feat(service): enforce three-workflow governance and instance lifecycle

Wire catalog policy, v2 manifests, stage access, activation metadata fixes, and coordinator instance advance around frozen workflow snapshots.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-13 02:50:23 +00:00
committed by weichen
co-authored by Cursor
parent 543ca310da
commit f07178fa73
25 changed files with 1163 additions and 297 deletions
+41 -18
View File
@@ -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}
}
+1 -1
View File
@@ -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"
+15 -7
View File
@@ -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
+6 -1
View File
@@ -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
}
+16 -9
View File
@@ -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 {
+169 -16
View File
@@ -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 &copyInst, 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 &copyInst, 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
}
+43
View File
@@ -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
}
+41 -39
View File
@@ -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"`
}
+73 -43
View File
@@ -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,