diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 43e07d2c..642512e0 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -19,6 +19,7 @@ import ( "chat-agent-system/internal/docgov" "chat-agent-system/internal/freedoc" "chat-agent-system/internal/kb" + "chat-agent-system/internal/projectcontext" "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/squad" "chat-agent-system/internal/storage" @@ -212,7 +213,7 @@ var SupportedStages = []StageMeta{ Stage: StageV05_Feasibility, Title: "V0.5 方案成型与法规对齐", Description: "法规与注册路径初筛、竞品调研、系统初步架构设计与关键选型建议", - Tasks: []string{"research", "feasibility_reg", "prelim_design"}, + Tasks: []string{"research", "feasibility_dev", "feasibility_reg", "user_requirements", "prelim_design"}, }, { Stage: StageV10_Quoting, @@ -346,36 +347,46 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun ErrorReason: docErr.Error(), }, docErr } - // 智能测试经理前置检查:必须依赖同项目已批准的系统需求规格书 (system_reqs) + // 智能测试工程师:统一项目上下文快照;SRS 仅为可选高权威来源 var upstreamSRDoc *docgov.DocumentRecord - if req.TaskKey == "test_manager" { - upstreamDocID := fmt.Sprintf("doc_proj%d_system_reqs", req.ProjectID) - if s.docGov != nil && req.ProjectID > 0 { - if gateErr := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, "system_reqs"); gateErr == nil { - doc, err := s.docGov.GetDocument(ctx, upstreamDocID) - if err == nil && doc != nil && doc.CurrentState == docgov.StateApproved { - upstreamSRDoc = doc + if req.TaskKey == "test_manager" || req.TaskKey == "system_reqs" { + priorDocs := make([]squad.PriorDocItem, 0) + facts := map[string]string{} + if req.Fields != nil { + for k, v := range req.Fields { + facts[k] = v + } + } + if s.ctxMgr != nil && req.ProjectID > 0 { + if coll, err := s.getProjectContextForUser(ctx, req.ProjectID, req.UserID); err == nil && coll != nil { + for k, v := range coll.BlackboardFacts { + facts[k] = v } } } - if upstreamSRDoc == nil { - return &TaskRunResult{ - TaskKey: req.TaskKey, - ArtifactTitle: artifactTitle, - WorkDir: req.WorkDir, - Success: false, - ErrorReason: "智能测试经理工作流前置条件未满足:必须先完成同项目的《系统需求规格书》并经批准 (approved) 后方可启动系统验证编制", - }, projectworkflow.ErrPrerequisiteNotMet + upstreamDocID := fmt.Sprintf("doc_proj%d_system_reqs", req.ProjectID) + if s.docGov != nil && req.ProjectID > 0 { + if doc, err := s.docGov.GetDocument(ctx, upstreamDocID); err == nil && doc != nil && doc.CurrentState == docgov.StateApproved { + upstreamSRDoc = doc + if doc.WorkDir != "" { + contentPath := filepath.Join(doc.WorkDir, "output.md") + if b, readErr := os.ReadFile(contentPath); readErr == nil && len(b) > 0 { + priorDocs = append(priorDocs, squad.PriorDocItem{ + DocKey: "system_reqs", Title: "系统需求规格书", Content: string(b), Stage: "approved", + }) + } + } + } } - - // 自动将已批准系统需求规格书注入为测试方案编制的依据输入 - if len(materials) == 0 && upstreamSRDoc.WorkDir != "" { - contentPath := filepath.Join(upstreamSRDoc.WorkDir, "output.md") - if b, err := os.ReadFile(contentPath); err == nil && len(b) > 0 { - materials = append(materials, squad.MaterialItem{ - Filename: "upstream_system_requirements.md", - Content: b, - }) + ctxSvc := projectcontext.NewService(s.pool) + snap, _ := ctxSvc.BuildSnapshot(ctx, projectcontext.BuildInput{ + ProjectID: req.ProjectID, TaskKey: req.TaskKey, UserID: req.UserID, + ProjectName: req.ProjectName, Facts: facts, Materials: materials, PriorDocs: priorDocs, + Knowledge: hits, + }) + if snap != nil { + if extra := projectcontext.MaterialsForWorkspace(snap, 12); len(extra) > 0 { + materials = append(materials, extra...) } } } diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 0f7a2c4c..9bd43fe2 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -8,6 +8,7 @@ import ( "os" "path/filepath" "regexp" + "strconv" "strings" "time" @@ -17,6 +18,7 @@ import ( "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/stream" ) @@ -63,11 +65,19 @@ type CoordinationResult struct { // CoordinatorEngine L2 协调引擎 type CoordinatorEngine struct { bizdevSvc bizdev.BizdevService + instSvc *projectinstance.Service assembler *ContextAssembler dialogueBackend agent.Backend dataDir string } +// SetInstanceService 注入项目工作流实例服务(对话推进与实例门禁合流)。 +func (e *CoordinatorEngine) SetInstanceService(svc *projectinstance.Service) { + if e != nil { + e.instSvc = svc + } +} + func NewCoordinatorEngine(bizdevSvc bizdev.BizdevService) *CoordinatorEngine { return &CoordinatorEngine{ bizdevSvc: bizdevSvc, @@ -670,6 +680,26 @@ 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 { + 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 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 + } + } + } + targetStageStr := cand.FilledSlots[intent.SlotTargetStage] targetStage := bizdev.MaturityStage(targetStageStr) if targetStage == "" { diff --git a/internal/coordinator/instance_advance.go b/internal/coordinator/instance_advance.go new file mode 100644 index 00000000..2c95b5ed --- /dev/null +++ b/internal/coordinator/instance_advance.go @@ -0,0 +1,53 @@ +package coordinator + +import ( + "context" + "fmt" + + "chat-agent-system/internal/intent" + "chat-agent-system/internal/projectinstance" +) + +func (e *CoordinatorEngine) tryAdvanceWorkflowInstance(ctx context.Context, projectID int64, userID int64) (*CoordinationResult, bool, error) { + if e == nil || e.instSvc == nil || !e.instSvc.Available() { + return nil, false, nil + } + live, err := e.instSvc.GetSettings(ctx, projectID, userID, true) + if err != nil || live == nil || live.LiveInstance == nil { + return nil, false, nil + } + stageKey := live.ActiveStageKey + if stageKey == "" { + return nil, false, nil + } + readiness, err := e.instSvc.GetStageReadiness(ctx, live.LiveInstance.ID, stageKey, userID, true) + if err != nil { + return nil, true, err + } + if !readiness.Ready { + return &CoordinationResult{ + Status: StatusRejected, + Intent: intent.IntentAdvanceStage, + RejectReason: "STAGE_GATE_NOT_READY", + NaturalReply: fmt.Sprintf("当前阶段尚未满足推进条件,仍有 %d 项阻断。", len(readiness.BlockingIssues)), + }, true, projectinstance.ErrStageGateNotReady + } + result, err := e.instSvc.AdvanceStage(ctx, live.LiveInstance.ID, userID, true, projectinstance.AdvanceStageInput{ + StageKey: stageKey, + ExpectedConfigRevision: readiness.ConfigRevision, + }) + if err != nil { + return nil, true, err + } + reply := fmt.Sprintf("已确认推进阶段【%s】。", stageKey) + if result.InstanceDone { + reply = "工作流已完成。您现在可以配置下一个工作流。" + } else if result.NextStageKey != "" { + reply = fmt.Sprintf("已确认推进至下一阶段【%s】。", result.NextStageKey) + } + return &CoordinationResult{ + Status: StatusAccepted, + Intent: intent.IntentAdvanceStage, + NaturalReply: reply, + }, true, nil +} diff --git a/internal/projectcontext/service.go b/internal/projectcontext/service.go new file mode 100644 index 00000000..562b5504 --- /dev/null +++ b/internal/projectcontext/service.go @@ -0,0 +1,280 @@ +package projectcontext + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/squad" +) + +const ( + AuthorityConfirmedFact = "confirmed_fact" + AuthorityApprovedDoc = "approved_document" + AuthorityUpload = "upload_material" + AuthorityMilestone = "milestone" + AuthorityDialogue = "dialogue" + AuthorityKnowledge = "knowledge" +) + +// Source 来源清单条目(全量枚举,按任务检索后分级注入)。 +type Source struct { + Type string `json:"type"` + ID string `json:"id"` + Version string `json:"version,omitempty"` + ContentHash string `json:"content_hash"` + AuthorityLevel string `json:"authority_level"` + PermissionScope string `json:"permission_scope"` + Referenced bool `json:"referenced"` + Snippet string `json:"snippet,omitempty"` +} + +// 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"` +} + +// BuildInput 构建快照所需输入。 +type BuildInput struct { + ProjectID int64 + TaskKey string + UserID int64 + ProjectName string + Facts map[string]string + Materials []squad.MaterialItem + PriorDocs []squad.PriorDocItem + Knowledge []squad.KnowledgeHit + Dialogue []string + Decisions []string +} + +// Service 项目上下文快照服务。 +type Service struct { + pool *pgxpool.Pool +} + +func NewService(pool *pgxpool.Pool) *Service { + return &Service{pool: pool} +} + +func (s *Service) Available() bool { + return s != nil && s.pool != nil +} + +// BuildSnapshot 建立来源清单并按任务检索相关片段(不 dump 全量原文)。 +func (s *Service) BuildSnapshot(ctx context.Context, in BuildInput) (*Snapshot, error) { + cutoff := time.Now() + sources := 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, + } + if !s.Available() || in.ProjectID <= 0 { + return snap, nil + } + + var id int64 + err := s.pool.QueryRow(ctx, ` + INSERT INTO project_context_snapshots (project_id, task_key, cutoff_at, aggregate_hash, source_count, created_by, frozen) + VALUES ($1, $2, $3, $4, $5, $6, true) RETURNING id + `, in.ProjectID, in.TaskKey, cutoff, aggHash, len(sources), in.UserID).Scan(&id) + if err != nil { + return snap, err + } + snap.ID = id + for _, src := range sources { + _, _ = s.pool.Exec(ctx, ` + INSERT INTO project_context_snapshot_sources + (snapshot_id, source_type, source_id, source_version, content_hash, authority_level, permission_scope, referenced) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (snapshot_id, source_type, source_id, source_version) DO NOTHING + `, id, src.Type, src.ID, src.Version, src.ContentHash, src.AuthorityLevel, src.PermissionScope, src.Referenced) + } + return snap, nil +} + +// MaterialsForWorkspace 按权威等级与任务目标注入工作区材料(核心固定 + 相关分批)。 +func MaterialsForWorkspace(snap *Snapshot, maxItems int) []squad.MaterialItem { + if snap == nil || len(snap.Sources) == 0 { + return nil + } + if maxItems <= 0 { + maxItems = 12 + } + out := make([]squad.MaterialItem, 0, maxItems) + for _, src := range snap.Sources { + if !src.Referenced || strings.TrimSpace(src.Snippet) == "" { + continue + } + out = append(out, squad.MaterialItem{ + Filename: fmt.Sprintf("%s_%s.md", src.Type, src.ID), + Content: []byte(src.Snippet), + }) + if len(out) >= maxItems { + break + } + } + return out +} + +func enumerateSources(in BuildInput) []Source { + out := make([]Source, 0) + for k, v := range in.Facts { + if strings.TrimSpace(v) == "" { + continue + } + out = append(out, Source{ + Type: AuthorityConfirmedFact, ID: k, ContentHash: hashText(v), + AuthorityLevel: AuthorityConfirmedFact, PermissionScope: "project", + }) + } + for i, m := range in.Materials { + body := string(m.Content) + if len(body) == 0 { + continue + } + id := m.Filename + if id == "" { + id = fmt.Sprintf("material_%d", i) + } + out = append(out, Source{ + Type: "upload_material", ID: id, ContentHash: hashText(body), + AuthorityLevel: AuthorityUpload, PermissionScope: "project", Snippet: truncate(body, 4000), + }) + } + for _, d := range in.PriorDocs { + body := d.Content + if body == "" { + continue + } + id := d.DocKey + if id == "" { + id = d.Title + } + out = append(out, Source{ + Type: "document", ID: id, Version: d.Stage, ContentHash: hashText(body), + AuthorityLevel: AuthorityApprovedDoc, PermissionScope: "project", Snippet: truncate(body, 4000), + }) + } + for i, line := range in.Dialogue { + if strings.TrimSpace(line) == "" { + 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), + }) + } + for i, d := range in.Decisions { + out = append(out, Source{ + Type: "decision", ID: fmt.Sprintf("decision_%d", i), ContentHash: hashText(d), + AuthorityLevel: AuthorityConfirmedFact, PermissionScope: "project", Snippet: truncate(d, 2000), + }) + } + for _, h := range in.Knowledge { + body := h.Content + if body == "" { + body = h.Title + } + out = append(out, Source{ + Type: "knowledge", ID: h.CiteKey, ContentHash: hashText(body), + AuthorityLevel: AuthorityKnowledge, PermissionScope: "global_kb", Snippet: truncate(body, 2000), + }) + } + sort.Slice(out, func(i, j int) bool { + return authorityRank(out[i].AuthorityLevel) < authorityRank(out[j].AuthorityLevel) + }) + return out +} + +func selectForTask(taskKey string, sources []Source) []Source { + if len(sources) == 0 { + return sources + } + keywords := taskKeywords(taskKey) + for i := range sources { + src := &sources[i] + if src.AuthorityLevel == AuthorityConfirmedFact || src.AuthorityLevel == AuthorityApprovedDoc { + src.Referenced = true + continue + } + if len(keywords) == 0 { + src.Referenced = src.AuthorityLevel != AuthorityDialogue + continue + } + hay := strings.ToLower(src.Snippet + " " + src.ID + " " + src.Type) + for _, kw := range keywords { + if strings.Contains(hay, kw) { + src.Referenced = true + break + } + } + } + return sources +} + +func taskKeywords(taskKey string) []string { + switch taskKey { + case "system_reqs": + return []string{"需求", "requirement", "功能", "system_reqs"} + case "test_manager": + return []string{"测试", "验证", "风险", "test", "verification", "requirement"} + default: + return nil + } +} + +func authorityRank(level string) int { + switch level { + case AuthorityConfirmedFact: + return 0 + case AuthorityApprovedDoc: + return 1 + case AuthorityUpload: + return 2 + case AuthorityMilestone: + return 3 + case AuthorityDialogue: + return 4 + default: + return 5 + } +} + +func aggregateHash(sources []Source) string { + raw, _ := json.Marshal(sources) + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} + +func hashText(s string) string { + sum := sha256.Sum256([]byte(s)) + return hex.EncodeToString(sum[:]) +} + +func truncate(s string, max int) string { + if len(s) <= max { + return s + } + return s[:max] + "\n…(truncated)" +} diff --git a/internal/projectinstance/errors.go b/internal/projectinstance/errors.go index 36036846..13147ace 100644 --- a/internal/projectinstance/errors.go +++ b/internal/projectinstance/errors.go @@ -1,24 +1,47 @@ package projectinstance -import "errors" +import ( + "errors" + + "chat-agent-system/internal/workflowdef" +) var ( - ErrNotAvailable = errors.New("project instance store not available") - ErrProjectNotFound = errors.New("project not found") - ErrForbiddenNotCreator = errors.New("forbidden: only project creator can configure workflow") - ErrPlanFrozen = errors.New("plan frozen: workflow already active") - ErrWorkflowAlreadyActive = errors.New("workflow already active for project") - ErrPlanChangedReconfirm = errors.New("plan changed: reconfirm required") - ErrInvalidPlan = errors.New("invalid workflow plan") - ErrValidationFailed = errors.New("workflow plan validation failed") - ErrTokenInvalid = errors.New("activation token invalid or expired") - ErrConcurrentConflict = errors.New("conflict: project version mismatch") - ErrPlanRevisionConflict = errors.New("conflict: plan revision mismatch") - ErrWorkflowVersionNotFound = errors.New("workflow version not found or not published") - ErrLegacySwitchBlocked = errors.New("legacy workflow switch blocked: use project settings activation") - ErrInstanceNotFound = errors.New("workflow instance not found") - ErrStageGateNotReady = errors.New("stage gate not ready: required documents not approved") - ErrStageNotActive = errors.New("stage is not the active stage") - ErrInvalidStage = errors.New("invalid stage key for instance") + ErrNotAvailable = errors.New("project instance store not available") + ErrProjectNotFound = errors.New("project not found") + ErrForbiddenNotCreator = errors.New("forbidden: only project creator can configure workflow") + ErrPlanFrozen = errors.New("plan frozen: workflow already active") + ErrWorkflowAlreadyActive = errors.New("workflow already active for project") + ErrPlanChangedReconfirm = errors.New("plan changed: reconfirm required") + ErrInvalidPlan = errors.New("invalid workflow plan") + ErrValidationFailed = errors.New("workflow plan validation failed") + ErrTokenInvalid = errors.New("activation token invalid or expired") + ErrConcurrentConflict = errors.New("conflict: project version mismatch") + ErrPlanRevisionConflict = errors.New("conflict: plan revision mismatch") + ErrWorkflowVersionNotFound = errors.New("workflow version not found or not published") + ErrLegacySwitchBlocked = errors.New("legacy workflow switch blocked: use project settings activation") + ErrInstanceNotFound = errors.New("workflow instance not found") + ErrStageGateNotReady = errors.New("stage gate not ready: required documents not approved") + ErrStageNotActive = errors.New("stage is not the active stage") + ErrInvalidStage = errors.New("invalid stage key for instance") ErrInstanceRevisionConflict = errors.New("conflict: instance config revision mismatch") + ErrEmptyWorkflow = errors.New("empty workflow: at least one enabled document required to activate") + ErrTerminationNotAllowed = errors.New("termination not allowed for instance state") ) + +// ValidationError 保留具体校验问题,供 HTTP 层返回可操作的错误信息。 +type ValidationError struct { + Issues []workflowdef.ValidationIssue +} + +func (e *ValidationError) Error() string { + return ErrValidationFailed.Error() +} + +func (e *ValidationError) Unwrap() error { + return ErrValidationFailed +} + +func newValidationError(result workflowdef.ValidationResult) error { + return &ValidationError{Issues: result.Issues} +} diff --git a/internal/projectinstance/gate.go b/internal/projectinstance/gate.go index f5d95f1a..5eefd7c5 100644 --- a/internal/projectinstance/gate.go +++ b/internal/projectinstance/gate.go @@ -25,7 +25,7 @@ func evaluateStageReadiness( continue } td, hasDef := taskDefMap[dt.TaskKey] - required := !hasDef || td.Required || td.Blocking + required := !hasDef || td.BlocksWhenEnabled() state := docStates[dt.TaskKey] if state == "" { state = "missing" diff --git a/internal/projectinstance/service.go b/internal/projectinstance/service.go index d442b14e..2fa513be 100644 --- a/internal/projectinstance/service.go +++ b/internal/projectinstance/service.go @@ -95,16 +95,20 @@ func (s *Service) GetSettings(ctx context.Context, projectID int64, userID int64 defs, listErr := s.wfStore.ListDefinitions(ctx) if listErr == nil { for _, d := range defs { - if d.LatestPublishedVersion <= 0 { + if d.LatestPublishedVersion <= 0 || !workflowdef.IsOfficialWorkflowKey(d.Key) { continue } ver, verErr := s.wfStore.GetVersion(ctx, d.ID, d.LatestPublishedVersion) if verErr != nil || ver == nil { continue } + name := d.Name + if name == "" || name == d.Key { + name = workflowdef.OfficialWorkflowName(d.Key) + } settings.PublishedWorkflows = append(settings.PublishedWorkflows, PublishedOption{ DefinitionKey: d.Key, - DefinitionName: d.Name, + DefinitionName: name, LatestVersion: d.LatestPublishedVersion, WorkflowVersion: ver.ID, ProductDomain: d.ProductDomain, @@ -147,10 +151,11 @@ func (s *Service) SaveWorkflowPlan(ctx context.Context, projectID int64, userID } validation := validatePlan(in.Plan, version) if !validation.Valid { - return nil, fmt.Errorf("%w: %d issues", ErrValidationFailed, len(validation.Issues)) + return nil, newValidationError(validation) } in.Plan.WorkflowVersionID = version.ID + applyDefinitionLabels(&in.Plan, version) planHash, err := computePlanHash(in.Plan) if err != nil { return nil, err @@ -197,9 +202,9 @@ func (s *Service) ActivationPreview(ctx context.Context, projectID int64, userID if err != nil { return nil, err } - validation := validatePlan(planInst.ConfigSnapshot, version) + validation := validateActivationPlan(planInst.ConfigSnapshot, version) if !validation.Valid { - return nil, fmt.Errorf("%w", ErrValidationFailed) + return nil, newValidationError(validation) } token, err := randomToken(24) @@ -264,12 +269,12 @@ func (s *Service) Activate(ctx context.Context, projectID int64, userID int64, i if err != nil { return nil, err } - validation := validatePlan(planInst.ConfigSnapshot, version) + validation := validateActivationPlan(planInst.ConfigSnapshot, version) if !validation.Valid { return nil, ErrValidationFailed } - activated, err := s.store.ActivateInstance(ctx, projectID, planInst, version.WorkflowKey, selVer, userID) + activated, err := s.store.ActivateInstance(ctx, projectID, planInst, version, selVer, userID) if err != nil { return nil, err } @@ -306,6 +311,9 @@ func (s *Service) BuildDefaultPlan(ctx context.Context, workflowKey string, vers if s.wfStore == nil { return nil, ErrNotAvailable } + if !workflowdef.IsOfficialWorkflowKey(workflowKey) { + return nil, ErrWorkflowVersionNotFound + } def, err := s.wfStore.GetDefinitionByKey(ctx, workflowKey) if err != nil { return nil, ErrWorkflowVersionNotFound diff --git a/internal/projectinstance/stage_access.go b/internal/projectinstance/stage_access.go index 89030780..b3d4b4e7 100644 --- a/internal/projectinstance/stage_access.go +++ b/internal/projectinstance/stage_access.go @@ -24,7 +24,12 @@ func (s *Service) ResolveStageAccess(ctx context.Context, projectID int64, userI } live, err := s.store.GetLiveInstance(ctx, projectID) - if err != nil || live == nil || live.Status == InstanceCompleted { + if err != nil || live == nil { + out.Restricted = true + return out, nil + } + if live.Status == InstanceCompleted || live.Status == "terminated" { + out.Restricted = true return out, nil } diff --git a/internal/projectinstance/stage_store.go b/internal/projectinstance/stage_store.go index 0d1af835..1942bdbe 100644 --- a/internal/projectinstance/stage_store.go +++ b/internal/projectinstance/stage_store.go @@ -117,16 +117,11 @@ func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey s if err != nil { return nil, err } - _, err = tx.Exec(ctx, ` - UPDATE project_doc_task_instances SET status = 'approved', updated_at = $2 - WHERE stage_instance_id = $1 AND enabled = true AND status <> 'skipped' - `, stageID, now) - if err != nil { - return nil, err - } - var nextStageID int64 var nextStageKey string + var projectID int64 + _ = tx.QueryRow(ctx, `SELECT project_id FROM project_workflow_instances WHERE id = $1`, instanceID).Scan(&projectID) + err = tx.QueryRow(ctx, ` SELECT id, stage_key FROM project_stage_instances WHERE workflow_instance_id = $1 AND enabled = true AND status = 'pending' @@ -134,12 +129,24 @@ func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey s `, instanceID).Scan(&nextStageID, &nextStageKey) if errors.Is(err, pgx.ErrNoRows) { _, err = tx.Exec(ctx, ` - UPDATE project_workflow_instances SET status = 'completed', config_revision = config_revision + 1, updated_at = $2 + UPDATE project_workflow_instances SET status = 'completed', config_revision = config_revision + 1, + completed_at = $2, completed_by = ( + SELECT activated_by FROM project_workflow_instances WHERE id = $1 + ), updated_at = $2 WHERE id = $1 `, instanceID, now) if err != nil { return nil, err } + if projectID > 0 { + _, err = tx.Exec(ctx, ` + UPDATE projects SET lifecycle_status = 'planning', active_workflow_key = '', updated_at = $2 + WHERE id = $1 + `, projectID, now) + if err != nil { + return nil, err + } + } } else if err != nil { return nil, err } else { diff --git a/internal/projectinstance/store.go b/internal/projectinstance/store.go index a7ab972e..942ce7c4 100644 --- a/internal/projectinstance/store.go +++ b/internal/projectinstance/store.go @@ -2,6 +2,8 @@ package projectinstance import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -10,6 +12,8 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/workflowdef" ) // TablesReady 检查 Phase 3 表是否已迁移。 @@ -37,7 +41,8 @@ type Store interface { GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) HasLiveInstance(ctx context.Context, projectID int64) (bool, error) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) - ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) + ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, 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 ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) @@ -252,11 +257,15 @@ func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst * return err } for _, dt := range st.DocTasks { + status := "pending" + if !dt.Enabled { + status = "skipped_disabled" + } var docTaskID int64 err = tx.QueryRow(ctx, ` INSERT INTO project_doc_task_instances (stage_instance_id, task_key, enabled, status) - VALUES ($1, $2, $3, 'pending') RETURNING id - `, stageID, dt.TaskKey, dt.Enabled).Scan(&docTaskID) + VALUES ($1, $2, $3, $4) RETURNING id + `, stageID, dt.TaskKey, dt.Enabled, status).Scan(&docTaskID) if err != nil { return err } @@ -282,7 +291,11 @@ func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst * return nil } -func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) { +func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) { + workflowKey := "" + if version != nil { + workflowKey = version.WorkflowKey + } tx, err := s.pool.Begin(ctx) if err != nil { return nil, err @@ -334,22 +347,99 @@ func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *W return nil, err } + if version != nil { + if err := s.applyRuntimeMetadata(ctx, tx, inst.ID, version.Definition); err != nil { + return nil, err + } + } + if err := tx.Commit(ctx); err != nil { return nil, err } return activated, 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 { + for _, dt := range st.DocTasks { + taskDefs[dt.Key] = dt + } + } + // 先完整读取并关闭结果集,再执行 UPDATE: + // pgx 在同一事务连接上存在未读完的 rows 时执行 Exec 会报 "conn busy"。 + rows, err := tx.Query(ctx, ` + SELECT dti.id, dti.task_key + FROM project_doc_task_instances dti + JOIN project_stage_instances si ON si.id = dti.stage_instance_id + WHERE si.workflow_instance_id = $1 + `, instanceID) + if err != nil { + return err + } + type taskInstanceRow struct { + id int64 + taskKey string + } + taskRows := make([]taskInstanceRow, 0) + for rows.Next() { + var r taskInstanceRow + if err := rows.Scan(&r.id, &r.taskKey); err != nil { + rows.Close() + return err + } + taskRows = append(taskRows, r) + } + iterErr := rows.Err() + rows.Close() + if iterErr != nil { + return iterErr + } + for _, r := range taskRows { + td, ok := taskDefs[r.taskKey] + if !ok { + continue + } + tplKey := td.TemplateAssetKey + tplVer := td.TemplateVersion + if td.Template != nil { + if td.Template.AssetKey != "" { + tplKey = td.Template.AssetKey + } + if td.Template.Version > 0 { + tplVer = td.Template.Version + } + } + if _, err := tx.Exec(ctx, ` + UPDATE project_doc_task_instances SET + executor_key = $2, quality_gate_key = $3, + template_asset_key = $4, template_version = $5, + output_doc_type = $6, output_doc_title = $7, updated_at = now() + WHERE id = $1 + `, r.id, td.ExecutorKey, td.QualityGateKey, tplKey, tplVer, td.OutputDocType, td.OutputDocTitle); err != nil { + return err + } + } + return nil +} + +func hashActivationToken(token string) string { + sum := sha256.Sum256([]byte(token)) + return hex.EncodeToString(sum[:]) +} + func (s *pgStore) SaveActivationToken(ctx context.Context, token string, projectID int64, planHash string, expectedVersion int, snapshot WorkflowPlan, createdBy int64, expiresAt time.Time) error { raw, _ := json.Marshal(snapshot) + tokenHash := hashActivationToken(token) _, err := s.pool.Exec(ctx, ` INSERT INTO workflow_activation_tokens (token, project_id, plan_hash, expected_project_version, config_snapshot, created_by, expires_at) VALUES ($1, $2, $3, $4, $5, $6, $7) - `, token, projectID, planHash, expectedVersion, raw, createdBy, expiresAt) + `, tokenHash, projectID, planHash, expectedVersion, raw, createdBy, expiresAt) return err } func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) { + tokenHash := hashActivationToken(token) var snapRaw []byte var storedHash string var usedAt *time.Time @@ -357,7 +447,7 @@ func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, proj err := s.pool.QueryRow(ctx, ` SELECT plan_hash, config_snapshot, used_at, expires_at FROM workflow_activation_tokens WHERE token = $1 AND project_id = $2 - `, token, projectID).Scan(&storedHash, &snapRaw, &usedAt, &expiresAt) + `, tokenHash, projectID).Scan(&storedHash, &snapRaw, &usedAt, &expiresAt) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrTokenInvalid } @@ -369,7 +459,7 @@ func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, proj } tag, err := s.pool.Exec(ctx, ` UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL - `, token) + `, tokenHash) if err != nil { return nil, err } @@ -381,6 +471,43 @@ func (s *pgStore) ConsumeActivationToken(ctx context.Context, token string, proj return &plan, nil } +func (s *pgStore) TerminateInstance(ctx context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + now := time.Now() + row := tx.QueryRow(ctx, ` + UPDATE project_workflow_instances SET status = 'terminated', terminated_at = $3, terminated_by = $4, + termination_reason = $5, config_revision = config_revision + 1, updated_at = $3 + WHERE id = $1 AND project_id = $2 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested') + RETURNING id, project_id, workflow_version_id, instance_no, status, config_revision, plan_hash, + config_snapshot, activated_by, activated_at, created_by, created_at, updated_at + `, instanceID, projectID, now, terminatedBy, reason) + inst, err := scanInstance(row) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrTerminationNotAllowed + } + if err != nil { + return nil, err + } + + _, err = tx.Exec(ctx, ` + UPDATE projects SET lifecycle_status = 'planning', active_workflow_key = '', updated_at = $2 + WHERE id = $1 + `, projectID, now) + if err != nil { + return nil, err + } + + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return inst, nil +} + // ============================================================================= // 内存 Store(单元测试 / 无 PG 迁移) // ============================================================================= @@ -579,7 +706,11 @@ func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInsta return ©Inst, nil } -func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, workflowKey string, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) { +func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) { + workflowKey := "" + if version != nil { + workflowKey = version.WorkflowKey + } m.mu.Lock() defer m.mu.Unlock() pm := m.ensureMeta(projectID) @@ -649,11 +780,6 @@ func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey st return nil, ErrInvalidStage } records[activeIdx].status = StageCompleted - for j := range records[activeIdx].tasks { - if records[activeIdx].tasks[j].enabled { - records[activeIdx].tasks[j].status = "approved" - } - } nextIdx := -1 for i, rec := range records { if rec.enabled && rec.status == StagePending { @@ -665,6 +791,11 @@ func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey st inst.UpdatedAt = time.Now() if nextIdx < 0 { inst.Status = InstanceCompleted + if pm := m.meta[inst.ProjectID]; pm.lifecycle != "" { + pm.lifecycle = LifecyclePlanning + pm.activeKey = "" + m.meta[inst.ProjectID] = pm + } } else { records[nextIdx].status = StageActive } @@ -693,10 +824,31 @@ func (m *memStore) MarkDocTaskApproved(_ context.Context, projectID int64, taskK return nil } +func (m *memStore) TerminateInstance(_ context.Context, instanceID int64, projectID int64, terminatedBy int64, reason string) (*WorkflowInstance, error) { + m.mu.Lock() + defer m.mu.Unlock() + inst := m.live[projectID] + if inst == nil || inst.ID != instanceID { + return nil, ErrInstanceNotFound + } + now := time.Now() + inst.Status = "terminated" + copyInst := *inst + delete(m.live, projectID) + pm := m.ensureMeta(projectID) + pm.lifecycle = LifecyclePlanning + pm.activeKey = "" + m.meta[projectID] = pm + _ = terminatedBy + _ = reason + _ = now + return ©Inst, 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() - m.tokens[token] = activationTokenRec{ + m.tokens[hashActivationToken(token)] = activationTokenRec{ projectID: projectID, planHash: planHash, expectedVer: expectedVersion, snapshot: snapshot, expiresAt: expiresAt, } @@ -706,12 +858,13 @@ func (m *memStore) SaveActivationToken(_ context.Context, token string, projectI func (m *memStore) ConsumeActivationToken(_ context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) { m.mu.Lock() defer m.mu.Unlock() - rec, ok := m.tokens[token] + tokenHash := hashActivationToken(token) + rec, ok := m.tokens[tokenHash] if !ok || rec.used || rec.projectID != projectID || rec.planHash != planHash || time.Now().After(rec.expiresAt) { return nil, ErrTokenInvalid } rec.used = true - m.tokens[token] = rec + m.tokens[tokenHash] = rec plan := rec.snapshot return &plan, nil } diff --git a/internal/projectinstance/terminate.go b/internal/projectinstance/terminate.go new file mode 100644 index 00000000..fffcaeb0 --- /dev/null +++ b/internal/projectinstance/terminate.go @@ -0,0 +1,43 @@ +package projectinstance + +import "context" + +// TerminateInput 受控终止输入。 +type TerminateInput struct { + Reason string `json:"reason"` + Confirm bool `json:"confirm"` + ExpectedConfigRevision int `json:"expected_config_revision"` +} + +// TerminateResult 终止结果。 +type TerminateResult struct { + Instance WorkflowInstance `json:"instance"` + Reason string `json:"reason"` +} + +func (s *Service) TerminateInstance(ctx context.Context, instanceID int64, projectID int64, userID int64, isCreator bool, in TerminateInput) (*TerminateResult, error) { + if !isCreator { + return nil, ErrForbiddenNotCreator + } + if !in.Confirm { + return nil, ErrTerminationNotAllowed + } + if in.Reason == "" { + return nil, ErrValidationFailed + } + 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 in.ExpectedConfigRevision > 0 && live.ConfigRevision != in.ExpectedConfigRevision { + return nil, ErrInstanceRevisionConflict + } + inst, err := s.store.TerminateInstance(ctx, instanceID, projectID, userID, in.Reason) + if err != nil { + return nil, err + } + return &TerminateResult{Instance: *inst, Reason: in.Reason}, nil +} diff --git a/internal/projectinstance/types.go b/internal/projectinstance/types.go index c317c027..595f1736 100644 --- a/internal/projectinstance/types.go +++ b/internal/projectinstance/types.go @@ -27,16 +27,18 @@ type DocDuty struct { // DocTaskPlan 文档任务裁剪与职责。 type DocTaskPlan struct { - TaskKey string `json:"task_key"` - Enabled bool `json:"enabled"` - Duties DocDuty `json:"duties"` + TaskKey string `json:"task_key"` + TaskTitle string `json:"task_title,omitempty"` + Enabled bool `json:"enabled"` + Duties DocDuty `json:"duties"` } // StagePlan 阶段裁剪与文档任务。 type StagePlan struct { - StageKey string `json:"stage_key"` - Enabled bool `json:"enabled"` - DocTasks []DocTaskPlan `json:"doc_tasks"` + StageKey string `json:"stage_key"` + StageName string `json:"stage_name,omitempty"` + Enabled bool `json:"enabled"` + DocTasks []DocTaskPlan `json:"doc_tasks"` } // WorkflowPlan 项目工作流规划快照。 @@ -99,31 +101,31 @@ type SavePlanInput struct { // ActivationPreview 激活预览与确认令牌。 type ActivationPreview struct { - ConfirmationToken string `json:"confirmation_token"` - PlanHash string `json:"plan_hash"` - ExpectedProjectVersion int `json:"expected_project_version"` - ExpectedPlanRevision int `json:"expected_plan_revision"` - Summary PreviewSummary `json:"summary"` - Validation workflowdef.ValidationResult `json:"validation"` - ExpiresAt time.Time `json:"expires_at"` + ConfirmationToken string `json:"confirmation_token"` + PlanHash string `json:"plan_hash"` + ExpectedProjectVersion int `json:"expected_project_version"` + ExpectedPlanRevision int `json:"expected_plan_revision"` + Summary PreviewSummary `json:"summary"` + Validation workflowdef.ValidationResult `json:"validation"` + ExpiresAt time.Time `json:"expires_at"` } // PreviewSummary 激活影响摘要。 type PreviewSummary struct { - WorkflowName string `json:"workflow_name"` - WorkflowVersion int `json:"workflow_version"` - EnabledStages int `json:"enabled_stages"` - EnabledDocTasks int `json:"enabled_doc_tasks"` - AssignedDuties int `json:"assigned_duties"` - FirstStageKey string `json:"first_stage_key,omitempty"` - FirstStageName string `json:"first_stage_name,omitempty"` + WorkflowName string `json:"workflow_name"` + WorkflowVersion int `json:"workflow_version"` + EnabledStages int `json:"enabled_stages"` + EnabledDocTasks int `json:"enabled_doc_tasks"` + AssignedDuties int `json:"assigned_duties"` + FirstStageKey string `json:"first_stage_key,omitempty"` + FirstStageName string `json:"first_stage_name,omitempty"` } // ActivateInput 二次确认激活。 type ActivateInput struct { - ConfirmationToken string `json:"confirmation_token"` - ExpectedProjectVersion int `json:"expected_project_version"` - ExpectedPlanRevision int `json:"expected_plan_revision"` + ConfirmationToken string `json:"confirmation_token"` + ExpectedProjectVersion int `json:"expected_project_version"` + ExpectedPlanRevision int `json:"expected_plan_revision"` } // ActivateResult 激活结果。 @@ -166,8 +168,8 @@ type StageInstance struct { // InstanceDetail 工作流实例详情(含阶段树)。 type InstanceDetail struct { Instance WorkflowInstance `json:"instance"` - ActiveStageKey string `json:"active_stage_key,omitempty"` - Stages []StageInstance `json:"stages"` + ActiveStageKey string `json:"active_stage_key,omitempty"` + Stages []StageInstance `json:"stages"` } // GateIssue 阶段门禁阻断项。 @@ -188,14 +190,14 @@ type DocTaskReadiness struct { // StageReadiness 阶段推进就绪评估。 type StageReadiness struct { - InstanceID int64 `json:"instance_id"` - StageKey string `json:"stage_key"` - StageName string `json:"stage_name"` - Ready bool `json:"ready"` - CanAdvance bool `json:"can_advance"` - BlockingIssues []GateIssue `json:"blocking_issues"` - DocTasks []DocTaskReadiness `json:"doc_tasks"` - ConfigRevision int `json:"config_revision"` + InstanceID int64 `json:"instance_id"` + StageKey string `json:"stage_key"` + StageName string `json:"stage_name"` + Ready bool `json:"ready"` + CanAdvance bool `json:"can_advance"` + BlockingIssues []GateIssue `json:"blocking_issues"` + DocTasks []DocTaskReadiness `json:"doc_tasks"` + ConfigRevision int `json:"config_revision"` } // AdvanceStageInput 创建者确认推进。 @@ -206,10 +208,10 @@ type AdvanceStageInput struct { // AdvanceStageResult 阶段推进结果。 type AdvanceStageResult struct { - Instance WorkflowInstance `json:"instance"` - CompletedStage string `json:"completed_stage"` - CompletedStageInstanceID int64 `json:"completed_stage_instance_id,omitempty"` - CompletedReadinessRevision int `json:"completed_readiness_revision,omitempty"` - NextStageKey string `json:"next_stage_key,omitempty"` - InstanceDone bool `json:"instance_done"` + Instance WorkflowInstance `json:"instance"` + CompletedStage string `json:"completed_stage"` + CompletedStageInstanceID int64 `json:"completed_stage_instance_id,omitempty"` + CompletedReadinessRevision int `json:"completed_readiness_revision,omitempty"` + NextStageKey string `json:"next_stage_key,omitempty"` + InstanceDone bool `json:"instance_done"` } diff --git a/internal/projectinstance/validator.go b/internal/projectinstance/validator.go index 267ba113..8a02c720 100644 --- a/internal/projectinstance/validator.go +++ b/internal/projectinstance/validator.go @@ -21,6 +21,12 @@ func validatePlan(plan WorkflowPlan, version *workflowdef.Version) workflowdef.V Code: "WORKFLOW_NOT_PUBLISHED", Message: "只能绑定已发布的工作流版本", }) } + if !workflowdef.IsOfficialWorkflowKey(plan.WorkflowDefinitionKey) { + issues = append(issues, workflowdef.ValidationIssue{ + Code: "WORKFLOW_NOT_SELECTABLE", Field: "workflow_definition_key", + Message: "仅允许选择正式三工作流", + }) + } if plan.WorkflowDefinitionKey != version.WorkflowKey { issues = append(issues, workflowdef.ValidationIssue{ Code: "WORKFLOW_KEY_MISMATCH", Field: "workflow_definition_key", @@ -79,9 +85,9 @@ func validatePlan(plan WorkflowPlan, version *workflowdef.Version) workflowdef.V }) continue } - if (td.Required || td.Blocking) && !dt.Enabled { + if !dt.Enabled && !td.IsCreatorConfigurable() { issues = append(issues, workflowdef.ValidationIssue{ - Code: "REQUIRED_DOC_DISABLED", Field: dt.TaskKey, Message: "必需/阻断文档任务不能关闭", + Code: "REQUIRED_DOC_DISABLED", Field: dt.TaskKey, Message: "该文档任务不可由创建者关闭", }) } if dt.Enabled { @@ -102,45 +108,38 @@ func validatePlan(plan WorkflowPlan, version *workflowdef.Version) workflowdef.V }) } - // 反向完整性:定义中的必需阶段/任务必须在规划中出现且启用 - planStageMap := make(map[string]StagePlan, len(plan.Stages)) - for _, st := range plan.Stages { - planStageMap[st.StageKey] = st - } - for _, def := range version.Definition.Stages { - if def.Optional { - continue - } - ps, ok := planStageMap[def.Key] - if !ok || !ps.Enabled { - issues = append(issues, workflowdef.ValidationIssue{ - Code: "MISSING_REQUIRED_STAGE", Field: def.Key, - Message: "必需阶段未启用或缺失", - }) - continue - } - planTaskEnabled := map[string]bool{} - for _, dt := range ps.DocTasks { - if dt.Enabled { - planTaskEnabled[dt.TaskKey] = true - } - } - for _, td := range def.DocTasks { - if !(td.Required || td.Blocking) { - continue - } - if !planTaskEnabled[td.Key] { - issues = append(issues, workflowdef.ValidationIssue{ - Code: "MISSING_REQUIRED_TASK", Field: td.Key, - Message: "必需/阻断文档任务未启用或缺失", - }) - } - } - } - return workflowdef.ValidationResult{Valid: len(issues) == 0, Issues: issues} } +func validateActivationPlan(plan WorkflowPlan, version *workflowdef.Version) workflowdef.ValidationResult { + base := validatePlan(plan, version) + if !base.Valid { + return base + } + if countEnabledDocTasks(plan) == 0 { + base.Valid = false + base.Issues = append(base.Issues, workflowdef.ValidationIssue{ + Code: "EMPTY_WORKFLOW", Message: "至少启用一个文档任务才能激活工作流", + }) + } + return base +} + +func countEnabledDocTasks(plan WorkflowPlan) int { + n := 0 + for _, st := range plan.Stages { + if !st.Enabled { + continue + } + for _, dt := range st.DocTasks { + if dt.Enabled { + n++ + } + } + } + return n +} + func validateDuties(d DocDuty, taskKey string) []workflowdef.ValidationIssue { issues := make([]workflowdef.ValidationIssue, 0) if d.AuthorUserID <= 0 { @@ -188,14 +187,16 @@ func buildDefaultPlan(version *workflowdef.Version) WorkflowPlan { } for _, st := range version.Definition.Stages { sp := StagePlan{ - StageKey: st.Key, - Enabled: !st.Optional, - DocTasks: make([]DocTaskPlan, 0, len(st.DocTasks)), + StageKey: st.Key, + StageName: st.Name, + Enabled: !st.Optional, + DocTasks: make([]DocTaskPlan, 0, len(st.DocTasks)), } for _, dt := range st.DocTasks { sp.DocTasks = append(sp.DocTasks, DocTaskPlan{ - TaskKey: dt.Key, - Enabled: dt.DefaultEnabled || dt.Required || dt.Blocking, + TaskKey: dt.Key, + TaskTitle: dt.Title, + Enabled: dt.DefaultEnabled, }) } plan.Stages = append(plan.Stages, sp) @@ -203,6 +204,35 @@ func buildDefaultPlan(version *workflowdef.Version) WorkflowPlan { return plan } +// applyDefinitionLabels 使用已发布定义覆盖展示名称,避免客户端把自定义文本写入冻结快照。 +func applyDefinitionLabels(plan *WorkflowPlan, version *workflowdef.Version) { + if plan == nil || version == nil { + return + } + stageDefs := make(map[string]workflowdef.StageDef, len(version.Definition.Stages)) + for _, stage := range version.Definition.Stages { + stageDefs[stage.Key] = stage + } + for stageIdx := range plan.Stages { + stage := &plan.Stages[stageIdx] + stageDef, ok := stageDefs[stage.StageKey] + if !ok { + continue + } + stage.StageName = stageDef.Name + taskDefs := make(map[string]workflowdef.DocTaskDef, len(stageDef.DocTasks)) + for _, task := range stageDef.DocTasks { + taskDefs[task.Key] = task + } + for taskIdx := range stage.DocTasks { + task := &stage.DocTasks[taskIdx] + if taskDef, exists := taskDefs[task.TaskKey]; exists { + task.TaskTitle = taskDef.Title + } + } + } +} + func previewSummary(plan WorkflowPlan, version *workflowdef.Version) PreviewSummary { sum := PreviewSummary{ WorkflowVersion: plan.WorkflowVersion, diff --git a/internal/projectworkflow/service.go b/internal/projectworkflow/service.go index 35e65fb8..47db21ff 100644 --- a/internal/projectworkflow/service.go +++ b/internal/projectworkflow/service.go @@ -121,6 +121,11 @@ func (s *Service) ListProjects(ctx context.Context, userID int64) ([]*Project, e return s.store.ListProjectsForUser(ctx, userID) } +// ListAllProjects 查询全部项目,仅供已完成系统管理员鉴权的调用方使用。 +func (s *Service) ListAllProjects(ctx context.Context) ([]*Project, error) { + return s.store.ListAllProjects(ctx) +} + // IsProjectCreator 判断用户是否为指定项目的权威创建人 func (s *Service) IsProjectCreator(ctx context.Context, projectID int64, userID int64) (bool, error) { p, err := s.store.GetProjectByID(ctx, projectID) diff --git a/internal/projectworkflow/store.go b/internal/projectworkflow/store.go index 594a58e0..41aee3a2 100644 --- a/internal/projectworkflow/store.go +++ b/internal/projectworkflow/store.go @@ -17,6 +17,7 @@ type Store interface { CreateProject(ctx context.Context, p *Project) (*Project, error) GetProjectByID(ctx context.Context, id int64) (*Project, error) GetProjectByCode(ctx context.Context, code string) (*Project, error) + ListAllProjects(ctx context.Context) ([]*Project, error) ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error @@ -179,6 +180,38 @@ func (s *pgStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Pro return list, nil } +func (s *pgStore) ListAllProjects(ctx context.Context) ([]*Project, error) { + query := ` + SELECT p.id, p.code, p.name, p.customer_name, p.device_type, p.intended_use, p.target_market, + p.current_stage, p.environment, p.status, COALESCE(p.creator_user_id, 0), + COALESCE(p.lifecycle_status, 'planning'), COALESCE(p.active_workflow_key, ''), + COALESCE(p.workflow_selection_version, 1), COALESCE(p.workflow_plan_revision, 0), + p.active_workflow_changed_at, p.active_workflow_changed_by, p.created_at, p.updated_at + FROM projects p + ORDER BY p.id DESC + ` + rows, err := s.pool.Query(ctx, query) + if err != nil { + return nil, fmt.Errorf("list all projects failed: %w", err) + } + defer rows.Close() + + var list []*Project + for rows.Next() { + p := &Project{} + if err := rows.Scan( + &p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket, + &p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID, + &p.LifecycleStatus, &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, &p.WorkflowPlanRevision, + &p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt, + ); err != nil { + return nil, err + } + list = append(list, p) + } + return list, rows.Err() +} + func (s *pgStore) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) { query := ` SELECT id, project_id, workflow_key, status, enabled, created_at, updated_at @@ -660,6 +693,17 @@ func (m *memStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Pr return list, nil } +func (m *memStore) ListAllProjects(ctx context.Context) ([]*Project, error) { + m.mu.RLock() + defer m.mu.RUnlock() + list := make([]*Project, 0, len(m.projects)) + for _, p := range m.projects { + cpy := *p + list = append(list, &cpy) + } + return list, nil +} + func (m *memStore) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) { m.mu.RLock() defer m.mu.RUnlock() diff --git a/internal/workflow/registry.go b/internal/workflow/registry.go index 8b5cd66b..9481a38b 100644 --- a/internal/workflow/registry.go +++ b/internal/workflow/registry.go @@ -4,14 +4,16 @@ import "fmt" // ExecutorRegistry 执行器注册表(schema v2 发布校验与运行时解析)。 var ExecutorRegistry = map[string]struct{}{ - "bizdev.stage.v1": {}, - "squad.system_reqs.v1": {}, - "squad.test_manager.v1": {}, - "bizdev.intake.v1": {}, - "bizdev.research.v1": {}, - "bizdev.feasibility.v1": {}, - "bizdev.prelim_design.v1": {}, - "bizdev.cdo_quote.v1": {}, + "bizdev.stage.v1": {}, + "squad.system_reqs.v1": {}, + "squad.test_manager.v1": {}, + "bizdev.intake.v1": {}, + "bizdev.research.v1": {}, + "bizdev.feasibility.v1": {}, + "bizdev.feasibility_dev.v1": {}, + "bizdev.user_requirements.v1": {}, + "bizdev.prelim_design.v1": {}, + "bizdev.cdo_quote.v1": {}, } // QualityGateRegistry 确定性质量门禁注册表。 diff --git a/internal/workflowdef/bootstrap.go b/internal/workflowdef/bootstrap.go index aa34e495..f4ab5fbc 100644 --- a/internal/workflowdef/bootstrap.go +++ b/internal/workflowdef/bootstrap.go @@ -8,32 +8,32 @@ import ( "github.com/jackc/pgx/v5/pgxpool" - "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/storage" ) -func legacySingleStage(key, name, desc, stageKey, taskKey, docType, docTitle, upstream string) CreateDefinitionInput { - return CreateDefinitionInput{ - Key: key, - Name: name, - Description: desc, - ProductDomain: "medical_device", - Definition: VersionDefinition{ - SchemaVersion: 1, - RequiredUpstreamWorkflow: upstream, - Stages: []StageDef{{ - Key: stageKey, Name: stageKey, Position: 0, - DocTasks: []DocTaskDef{{ - Key: taskKey, Title: docTitle, Required: true, Blocking: true, DefaultEnabled: true, - OutputDocType: docType, OutputDocTitle: docTitle, - TemplateAssetKey: taskKey, TemplateVersion: 1, - }}, - }}, - }, - } +var officialSeedMeta = map[string]struct { + Name string + Description string + Domain string +}{ + OfficialKeyMain: { + Name: "商务对接流程", + Description: "V0.1 需求摸底 → V0.5 方案与可行性 → V1.0 商务报价(7 个受控文档)", + Domain: "medical_device", + }, + OfficialKeyProductManager: { + Name: "智能产品经理流程", + Description: "基于项目上下文独立生成系统需求规格书", + Domain: "medical_device", + }, + OfficialKeyTestManager: { + Name: "智能测试工程师流程", + Description: "基于项目上下文独立生成系统验证方案及计划", + Domain: "medical_device", + }, } -// BootstrapSeeds 幂等导入静态 catalog、首批行业工作流与 templates/docs。 +// BootstrapSeeds 幂等导入模板与三正式 v2 工作流;非正式定义标记为不可选。 func BootstrapSeeds(ctx context.Context, pool *pgxpool.Pool, cas *storage.CAS, templatesDir string, publishedBy int64) error { if pool == nil || !TablesReady(ctx, pool) { return nil @@ -60,80 +60,44 @@ func BootstrapSeeds(ctx context.Context, pool *pgxpool.Pool, cas *storage.CAS, t _, _ = svc.UploadTemplate(ctx, key, ent.Name(), data, ent.Name(), publishedBy) } - seeds := []CreateDefinitionInput{ - legacySingleStage("main", "商务与研发综合主线", "V0.1 分诊到 V0.6 阶段评审", "main", "main", "main_proposal", "商务研发综合方案", ""), - legacySingleStage("product_manager", "智能产品经理", "系统需求规格书生成", "system_reqs", "system_reqs", "system_reqs", "系统需求规格书", ""), - legacySingleStage("test_manager", "智能测试经理", "系统验证方案及计划", "system_test_plan", "test_manager", "system_test_plan", "系统验证方案及计划", "product_manager"), - medicalDeviceWorkflow(), - nonMedicalSoftwareWorkflow(), - } - for _, seed := range seeds { - if err := store.UpsertPublishedSeed(ctx, seed, publishedBy); err != nil { + for _, key := range OfficialWorkflowKeys() { + meta := officialSeedMeta[key] + filename := V2ManifestFiles[key] + def, loadErr := LoadManifestDefinition(filename) + if loadErr != nil { + return loadErr + } + normalizeManifestDefinition(&def) + if err := store.UpsertPublishedSeed(ctx, CreateDefinitionInput{ + Key: key, + Name: meta.Name, + Description: meta.Description, + ProductDomain: meta.Domain, + Definition: def, + }, publishedBy); err != nil { return err } } - return nil + + return store.RetireNonOfficialDefinitions(ctx) } -func medicalDeviceWorkflow() CreateDefinitionInput { - return CreateDefinitionInput{ - Key: "medical_device_development", - Name: "医疗器械研发", - Description: "商务摸底 → 需求策划 → 设计开发 → 验证确认 → 设计转移", - ProductDomain: "medical_device", - Definition: VersionDefinition{ - SchemaVersion: 1, - Stages: []StageDef{ - stage("biz_intake", "商务摸底", 0, doc("intake", "客户需求采集", "intake", "客户需求采集表")), - stage("req_planning", "需求与法规策划", 1, doc("system_reqs", "系统需求规格书", "system_reqs", "系统需求规格书")), - stage("design_dev", "设计开发", 2, doc("prelim_design", "初步设计", "prelim_design", "初步设计文档")), - stage("verification", "设计验证/确认", 3, doc("test_manager", "系统验证方案", "system_test_plan", "系统验证方案及计划")), - stage("transfer", "设计转移/发布", 4, doc("feasibility_reg", "法规可行性", "feasibility_reg", "法规可行性评估")), - }, - }, +func normalizeManifestDefinition(def *VersionDefinition) { + if def == nil { + return + } + def.RequiredUpstreamWorkflow = "" + for i := range def.Stages { + for j := range def.Stages[i].DocTasks { + dt := &def.Stages[i].DocTasks[j] + if dt.CreatorConfigurable || dt.BlockingWhenEnabled || dt.DefaultEnabled { + continue + } + if dt.Required || dt.Blocking { + dt.CreatorConfigurable = true + dt.BlockingWhenEnabled = true + dt.DefaultEnabled = true + } + } } } - -func nonMedicalSoftwareWorkflow() CreateDefinitionInput { - return CreateDefinitionInput{ - Key: "non_medical_software", - Name: "非医疗纯软件", - Description: "机会范围 → 需求 → 架构 → 实现 → 测试验收 → 发布复盘", - ProductDomain: "non_medical_software", - Definition: VersionDefinition{ - SchemaVersion: 1, - Stages: []StageDef{ - stage("opportunity", "机会与范围", 0, doc("intake", "范围说明", "scope_doc", "项目范围说明")), - stage("requirements", "产品需求", 1, doc("user_requirements", "用户需求", "user_requirements", "用户需求说明")), - stage("architecture", "架构与计划", 2, doc("feasibility_dev", "技术可行性", "feasibility_dev", "技术可行性分析")), - stage("implementation", "实现", 3, doc("prelim_design", "架构设计", "architecture_design", "架构设计文档")), - stage("testing", "测试与验收", 4, doc("test_manager", "测试计划", "test_plan", "测试计划")), - stage("release", "发布复盘", 5, doc("research", "发布复盘", "release_review", "发布复盘报告")), - }, - }, - } -} - -func stage(key, name string, pos int, tasks ...DocTaskDef) StageDef { - return StageDef{Key: key, Name: name, Position: pos, DocTasks: tasks} -} - -func doc(key, title, docType, docTitle string) DocTaskDef { - return DocTaskDef{ - Key: key, Title: title, Required: true, Blocking: true, DefaultEnabled: true, - TemplateAssetKey: key, TemplateVersion: 1, - OutputDocType: docType, OutputDocTitle: docTitle, - } -} - -// SeedFromMemoryCatalog 将内存 catalog 转为种子(测试用)。 -func SeedFromMemoryCatalog(cat projectworkflow.Catalog) []CreateDefinitionInput { - if cat == nil { - return nil - } - out := make([]CreateDefinitionInput, 0) - for _, d := range cat.List() { - out = append(out, legacySingleStage(d.Key, d.Name, d.Description, d.StageKey, d.TaskKey, d.OutputDocType, d.OutputDocTitle, d.RequiredUpstreamWorkflow)) - } - return out -} diff --git a/internal/workflowdef/catalog_policy.go b/internal/workflowdef/catalog_policy.go new file mode 100644 index 00000000..13f2edf0 --- /dev/null +++ b/internal/workflowdef/catalog_policy.go @@ -0,0 +1,53 @@ +package workflowdef + +// 正式项目可选工作流(唯一对外目录)。 +const ( + OfficialKeyMain = "main" + OfficialKeyProductManager = "product_manager" + OfficialKeyTestManager = "test_manager" +) + +var officialWorkflowKeys = map[string]struct{}{ + OfficialKeyMain: {}, + OfficialKeyProductManager: {}, + OfficialKeyTestManager: {}, +} + +var officialWorkflowNames = map[string]string{ + OfficialKeyMain: "商务对接流程", + OfficialKeyProductManager: "智能产品经理流程", + OfficialKeyTestManager: "智能测试工程师流程", +} + +var officialCatalogOrder = map[string]int{ + OfficialKeyMain: 10, + OfficialKeyProductManager: 20, + OfficialKeyTestManager: 30, +} + +// IsOfficialWorkflowKey 是否为三正式工作流 key。 +func IsOfficialWorkflowKey(key string) bool { + _, ok := officialWorkflowKeys[key] + return ok +} + +// OfficialWorkflowName 返回正式对外名称。 +func OfficialWorkflowName(key string) string { + if name, ok := officialWorkflowNames[key]; ok { + return name + } + return key +} + +// OfficialCatalogOrder 返回目录排序权重。 +func OfficialCatalogOrder(key string) int { + if order, ok := officialCatalogOrder[key]; ok { + return order + } + return 9999 +} + +// OfficialWorkflowKeys 返回三正式 key 列表(稳定顺序)。 +func OfficialWorkflowKeys() []string { + return []string{OfficialKeyMain, OfficialKeyProductManager, OfficialKeyTestManager} +} diff --git a/internal/workflowdef/catalog_store.go b/internal/workflowdef/catalog_store.go new file mode 100644 index 00000000..eb9bf1a9 --- /dev/null +++ b/internal/workflowdef/catalog_store.go @@ -0,0 +1,68 @@ +package workflowdef + +import "context" + +// ListProjectSelectableDefinitions 返回项目可选的已发布正式工作流(严格三项白名单)。 +func (s *Store) ListProjectSelectableDefinitions(ctx context.Context) ([]Definition, error) { + if !s.enabled() { + return nil, ErrNotAvailable + } + rows, err := s.pool.Query(ctx, ` + SELECT id, def_key, name, description, status, latest_published_version, product_domain, + COALESCE(project_selectable, false), COALESCE(catalog_order, 0), created_at, updated_at + FROM workflow_definitions + WHERE def_key = ANY($1) AND latest_published_version > 0 + ORDER BY COALESCE(catalog_order, 0) ASC, def_key ASC + `, OfficialWorkflowKeys()) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]Definition, 0, 3) + for rows.Next() { + var d Definition + if err := rows.Scan(&d.ID, &d.Key, &d.Name, &d.Description, &d.Status, &d.LatestPublishedVersion, &d.ProductDomain, + &d.ProjectSelectable, &d.CatalogOrder, &d.CreatedAt, &d.UpdatedAt); err != nil { + return nil, err + } + if !IsOfficialWorkflowKey(d.Key) { + continue + } + if !d.ProjectSelectable { + d.ProjectSelectable = true + } + if d.Name == "" || d.Name == d.Key { + d.Name = OfficialWorkflowName(d.Key) + } + out = append(out, d) + } + return out, rows.Err() +} + +// RetireNonOfficialDefinitions 将非三正式定义标记为不可选。 +func (s *Store) RetireNonOfficialDefinitions(ctx context.Context) error { + if !s.enabled() { + return ErrNotAvailable + } + _, err := s.pool.Exec(ctx, ` + UPDATE workflow_definitions + SET project_selectable = false, status = 'retired', updated_at = now() + WHERE def_key <> ALL($1) + `, OfficialWorkflowKeys()) + if err != nil { + return err + } + _, err = s.pool.Exec(ctx, ` + UPDATE workflow_definitions + SET project_selectable = true, + catalog_order = CASE def_key + WHEN 'main' THEN 10 + WHEN 'product_manager' THEN 20 + WHEN 'test_manager' THEN 30 + ELSE catalog_order + END, + updated_at = now() + WHERE def_key = ANY($1) + `, OfficialWorkflowKeys()) + return err +} diff --git a/internal/workflowdef/manifests/main.v2.json b/internal/workflowdef/manifests/main.v2.json index bf49508f..d6b78a76 100644 --- a/internal/workflowdef/manifests/main.v2.json +++ b/internal/workflowdef/manifests/main.v2.json @@ -5,46 +5,49 @@ "stages": [ { "key": "V0.1_triage", - "name": "闪电摸底与盲区探测", + "name": "V0.1 需求摸底", "position": 10, "optional": false, "doc_tasks": [{ "key": "intake", - "title": "客户需求采集", - "required": true, - "blocking": true, + "title": "用户初始需求收集表", "default_enabled": true, + "creator_configurable": true, + "blocking_when_enabled": true, "executor_key": "bizdev.intake.v1", "quality_gate_key": "quality.generic.v1", "output_doc_type": "intake", - "output_doc_title": "初始需求表" + "output_doc_title": "用户初始需求收集表" }] }, { "key": "V0.5_feasibility", - "name": "方案成型与法规对齐", + "name": "V0.5 方案与可行性", "position": 20, "optional": false, "doc_tasks": [ - {"key": "research", "title": "竞品调研", "required": true, "blocking": true, "default_enabled": true, "executor_key": "bizdev.research.v1", "output_doc_type": "research", "output_doc_title": "竞品调研报告"}, - {"key": "feasibility_reg", "title": "法规可行性", "required": true, "blocking": true, "default_enabled": true, "executor_key": "bizdev.feasibility.v1", "output_doc_type": "feasibility_reg", "output_doc_title": "法规可行性分析"}, - {"key": "prelim_design", "title": "初步设计", "required": true, "blocking": true, "default_enabled": true, "executor_key": "bizdev.prelim_design.v1", "output_doc_type": "prelim_design", "output_doc_title": "初步设计说明"} + {"key": "research", "title": "调研报告", "default_enabled": true, "creator_configurable": true, "blocking_when_enabled": true, "executor_key": "bizdev.research.v1", "quality_gate_key": "quality.generic.v1", "output_doc_type": "research", "output_doc_title": "调研报告"}, + {"key": "feasibility_dev", "title": "设计开发可行性评估报告", "default_enabled": true, "creator_configurable": true, "blocking_when_enabled": true, "executor_key": "bizdev.feasibility_dev.v1", "quality_gate_key": "quality.generic.v1", "output_doc_type": "feasibility_dev", "output_doc_title": "设计开发可行性评估报告"}, + {"key": "feasibility_reg", "title": "注册路径可行性评估报告", "default_enabled": true, "creator_configurable": true, "blocking_when_enabled": true, "executor_key": "bizdev.feasibility.v1", "quality_gate_key": "quality.generic.v1", "output_doc_type": "feasibility_reg", "output_doc_title": "注册路径可行性评估报告"}, + {"key": "user_requirements", "title": "用户需求表", "default_enabled": true, "creator_configurable": true, "blocking_when_enabled": true, "executor_key": "bizdev.user_requirements.v1", "quality_gate_key": "quality.generic.v1", "output_doc_type": "user_requirements", "output_doc_title": "用户需求表"}, + {"key": "prelim_design", "title": "系统初步设计方案", "default_enabled": true, "creator_configurable": true, "blocking_when_enabled": true, "executor_key": "bizdev.prelim_design.v1", "quality_gate_key": "quality.generic.v1", "output_doc_type": "prelim_design", "output_doc_title": "系统初步设计方案"} ] }, { "key": "V1.0_quoting", - "name": "工程拆解与商业锁定", + "name": "V1.0 商务报价", "position": 30, "optional": false, "doc_tasks": [{ "key": "cdo_quote", - "title": "CDO 报价与核算", - "required": true, - "blocking": true, + "title": "CDO 报价单", "default_enabled": true, + "creator_configurable": true, + "blocking_when_enabled": true, "executor_key": "bizdev.cdo_quote.v1", + "quality_gate_key": "quality.generic.v1", "output_doc_type": "cdo_quote", - "output_doc_title": "CDO 报价核算" + "output_doc_title": "CDO 报价单" }] } ] diff --git a/internal/workflowdef/manifests/product_manager.v2.json b/internal/workflowdef/manifests/product_manager.v2.json index 640ab211..ad87d2d4 100644 --- a/internal/workflowdef/manifests/product_manager.v2.json +++ b/internal/workflowdef/manifests/product_manager.v2.json @@ -10,9 +10,9 @@ "doc_tasks": [{ "key": "system_reqs", "title": "系统需求规格书", - "required": true, - "blocking": true, "default_enabled": true, + "creator_configurable": true, + "blocking_when_enabled": true, "executor_key": "squad.system_reqs.v1", "quality_gate_key": "quality.system_reqs.v2", "output_doc_type": "system_reqs", diff --git a/internal/workflowdef/manifests/test_manager.v2.json b/internal/workflowdef/manifests/test_manager.v2.json index 2a9bf385..a55e49f1 100644 --- a/internal/workflowdef/manifests/test_manager.v2.json +++ b/internal/workflowdef/manifests/test_manager.v2.json @@ -2,7 +2,6 @@ "schema_version": 2, "workflow_key": "test_manager", "product_domains": ["medical_device", "non_medical_software"], - "required_upstream_workflow": "product_manager", "stages": [{ "key": "system_verification_planning", "name": "系统验证策划", @@ -11,9 +10,9 @@ "doc_tasks": [{ "key": "test_manager", "title": "系统验证方案及计划", - "required": true, - "blocking": true, "default_enabled": true, + "creator_configurable": true, + "blocking_when_enabled": true, "executor_key": "squad.test_manager.v1", "quality_gate_key": "quality.test_manager.v2", "output_doc_type": "test_manager", diff --git a/internal/workflowdef/store.go b/internal/workflowdef/store.go index 22adf693..7743defd 100644 --- a/internal/workflowdef/store.go +++ b/internal/workflowdef/store.go @@ -361,17 +361,14 @@ func (s *Store) ListPublishedCatalogEntries(ctx context.Context) ([]Version, err return out, rows.Err() } -// UpsertPublishedSeed 幂等写入已发布种子(bootstrap)。 +// UpsertPublishedSeed 按内容 hash 幂等发布种子;已有版本但 hash 变化时递增新版本。 func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInput, publishedBy int64) error { if !s.enabled() { return ErrNotAvailable } - existing, err := s.GetDefinitionByKey(ctx, in.Key) - if err == nil && existing != nil && existing.LatestPublishedVersion > 0 { - return nil - } - if err != nil && !errors.Is(err, ErrNotFound) { - return err + validation := ValidateVersionBySchema(in.Definition) + if !validation.Valid { + return fmt.Errorf("seed validation failed: %d issues", len(validation.Issues)) } defHash, err := hashDefinition(in.Definition) @@ -380,6 +377,11 @@ func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInpu } defRaw, _ := json.Marshal(in.Definition) + existing, err := s.GetDefinitionByKey(ctx, in.Key) + if err != nil && !errors.Is(err, ErrNotFound) { + return err + } + tx, err := s.pool.Begin(ctx) if err != nil { return err @@ -387,13 +389,38 @@ func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInpu defer tx.Rollback(ctx) var defID int64 + nextVersion := 1 if existing != nil { defID = existing.ID + if existing.LatestPublishedVersion > 0 { + var latestHash string + _ = tx.QueryRow(ctx, ` + SELECT definition_hash FROM workflow_versions + WHERE workflow_id = $1 AND version = $2 AND status = 'published' + `, defID, existing.LatestPublishedVersion).Scan(&latestHash) + if latestHash == defHash { + _, _ = tx.Exec(ctx, ` + UPDATE workflow_definitions SET name = $2, description = $3, project_selectable = true, + catalog_order = $4, updated_at = now() + WHERE id = $1 + `, defID, in.Name, in.Description, OfficialCatalogOrder(in.Key)) + return tx.Commit(ctx) + } + nextVersion = existing.LatestPublishedVersion + 1 + } + _, err = tx.Exec(ctx, ` + UPDATE workflow_definitions SET name = $2, description = $3, product_domain = $4, + project_selectable = true, catalog_order = $5, updated_at = now() + WHERE id = $1 + `, defID, in.Name, in.Description, in.ProductDomain, OfficialCatalogOrder(in.Key)) + if err != nil { + return err + } } else { err = tx.QueryRow(ctx, ` - INSERT INTO workflow_definitions (def_key, name, description, product_domain, latest_published_version) - VALUES ($1, $2, $3, $4, 1) RETURNING id - `, in.Key, in.Name, in.Description, in.ProductDomain).Scan(&defID) + INSERT INTO workflow_definitions (def_key, name, description, product_domain, latest_published_version, project_selectable, catalog_order) + VALUES ($1, $2, $3, $4, 0, true, $5) RETURNING id + `, in.Key, in.Name, in.Description, in.ProductDomain, OfficialCatalogOrder(in.Key)).Scan(&defID) if err != nil { return err } @@ -401,16 +428,16 @@ func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInpu _, err = tx.Exec(ctx, ` INSERT INTO workflow_versions (workflow_id, version, status, schema_version, definition_json, definition_hash, published_by, published_at, created_by) - VALUES ($1, 1, 'published', $2, $3, $4, $5, now(), $5) + VALUES ($1, $2, 'published', $3, $4, $5, $6, now(), $6) ON CONFLICT (workflow_id, version) DO NOTHING - `, defID, in.Definition.SchemaVersion, defRaw, defHash, publishedBy) + `, defID, nextVersion, in.Definition.SchemaVersion, defRaw, defHash, publishedBy) if err != nil { return err } _, err = tx.Exec(ctx, ` - UPDATE workflow_definitions SET latest_published_version = 1, updated_at = now() WHERE id = $1 - `, defID) + UPDATE workflow_definitions SET latest_published_version = $2, updated_at = now() WHERE id = $1 + `, defID, nextVersion) if err != nil { return err } diff --git a/internal/workflowdef/types.go b/internal/workflowdef/types.go index 0072f275..8ca20247 100644 --- a/internal/workflowdef/types.go +++ b/internal/workflowdef/types.go @@ -20,11 +20,13 @@ type TemplateRef struct { // DocTaskDef 阶段内文档任务定义。 type DocTaskDef struct { - Key string `json:"key"` - Title string `json:"title"` - Required bool `json:"required"` - Blocking bool `json:"blocking"` - DefaultEnabled bool `json:"default_enabled"` + Key string `json:"key"` + Title string `json:"title"` + Required bool `json:"required"` + Blocking bool `json:"blocking"` + DefaultEnabled bool `json:"default_enabled"` + CreatorConfigurable bool `json:"creator_configurable"` + BlockingWhenEnabled bool `json:"blocking_when_enabled"` TemplateAssetKey string `json:"template_asset_key,omitempty"` TemplateVersion int `json:"template_version,omitempty"` Template *TemplateRef `json:"template,omitempty"` @@ -63,11 +65,29 @@ type Definition struct { Description string `json:"description"` Status string `json:"status"` ProductDomain string `json:"product_domain,omitempty"` + ProjectSelectable bool `json:"project_selectable"` + CatalogOrder int `json:"catalog_order"` LatestPublishedVersion int `json:"latest_published_version"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } +// IsCreatorConfigurable 文档是否可由创建者关闭。 +func (dt DocTaskDef) IsCreatorConfigurable() bool { + if dt.CreatorConfigurable { + return true + } + return !dt.Required && !dt.Blocking +} + +// BlocksWhenEnabled 启用后是否阻断阶段推进。 +func (dt DocTaskDef) BlocksWhenEnabled() bool { + if dt.BlockingWhenEnabled { + return true + } + return dt.Required || dt.Blocking +} + // Version 工作流版本记录。 type Version struct { ID int64 `json:"id"` diff --git a/internal/workflowdef/validator_v2.go b/internal/workflowdef/validator_v2.go index 612fd5a1..a5240f69 100644 --- a/internal/workflowdef/validator_v2.go +++ b/internal/workflowdef/validator_v2.go @@ -6,13 +6,26 @@ import ( "chat-agent-system/internal/workflow" ) +var officialDocCounts = map[string]int{ + OfficialKeyMain: 7, + OfficialKeyProductManager: 1, + OfficialKeyTestManager: 1, +} + // ValidateDefinitionV2 schema v2 发布校验。 func ValidateDefinitionV2(def VersionDefinition) ValidationResult { issues := make([]ValidationIssue, 0) if def.SchemaVersion < 2 { issues = append(issues, ValidationIssue{Code: "SCHEMA_VERSION", Message: "schema_version must be >= 2"}) } + if def.RequiredUpstreamWorkflow != "" { + issues = append(issues, ValidationIssue{ + Code: "UPSTREAM_NOT_ALLOWED", Message: "正式工作流不得声明 required_upstream_workflow", + }) + } seenPos := map[int]bool{} + seenTask := map[string]bool{} + docCount := 0 for _, st := range def.Stages { if st.Key == "" { issues = append(issues, ValidationIssue{Code: "STAGE_KEY_REQUIRED", Message: "stage key required"}) @@ -25,7 +38,19 @@ func ValidateDefinitionV2(def VersionDefinition) ValidationResult { issues = append(issues, ValidationIssue{Code: "DUPLICATE_POSITION", Field: st.Key, Message: "duplicate stage position"}) } seenPos[st.Position] = true + if len(st.DocTasks) == 0 { + issues = append(issues, ValidationIssue{Code: "STAGE_NO_DOC_TASKS", Field: st.Key, Message: "each stage must have at least one doc task"}) + } for _, dt := range st.DocTasks { + docCount++ + if dt.Key == "" { + issues = append(issues, ValidationIssue{Code: "DOC_TASK_KEY_REQUIRED", Message: "doc task key required"}) + continue + } + if seenTask[dt.Key] { + issues = append(issues, ValidationIssue{Code: "DUPLICATE_DOC_TASK", Field: dt.Key, Message: "duplicate doc task key"}) + } + seenTask[dt.Key] = true if dt.ExecutorKey != "" { if err := workflow.ValidateExecutorKey(dt.ExecutorKey); err != nil { issues = append(issues, ValidationIssue{Code: "UNKNOWN_EXECUTOR", Field: dt.Key, Message: err.Error()}) @@ -44,6 +69,17 @@ func ValidateDefinitionV2(def VersionDefinition) ValidationResult { if len(def.Stages) == 0 { issues = append(issues, ValidationIssue{Code: "NO_STAGES", Message: "at least one stage required"}) } + if def.WorkflowKey != "" && IsOfficialWorkflowKey(def.WorkflowKey) { + if want, ok := officialDocCounts[def.WorkflowKey]; ok && docCount != want { + issues = append(issues, ValidationIssue{ + Code: "OFFICIAL_DOC_COUNT", Field: def.WorkflowKey, + Message: fmt.Sprintf("official workflow %s requires exactly %d doc tasks, got %d", def.WorkflowKey, want, docCount), + }) + } + if name := OfficialWorkflowName(def.WorkflowKey); name == def.WorkflowKey { + // name check happens at publish layer via Definition.Name + } + } return ValidationResult{Valid: len(issues) == 0, Issues: issues} }