feat(workflow): reopen a completed stage without pausing the later one

Selected documents return to drafting, and advance asks the creator to acknowledge an open recheck mark.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-20 15:35:07 +00:00
co-authored by Cursor
parent 432884911f
commit 4a7b8cc26f
10 changed files with 437 additions and 21 deletions
+23
View File
@@ -991,6 +991,29 @@ func (e *GovernanceEngine) ApproverReject(ctx context.Context, docID string, act
return doc, nil
}
// ReopenForRevision moves an approved document back to drafting without recording a rejection.
func (e *GovernanceEngine) ReopenForRevision(ctx context.Context, docID string, actorID int64, actorName, reason string) (*DocumentRecord, error) {
e.mu.Lock()
defer e.mu.Unlock()
doc, exists := e.memDocs[docID]
if !exists {
var err error
doc, err = e.loadDocumentFromDBLocked(ctx, docID)
if err != nil || doc == nil {
return nil, ErrDocumentNotFound
}
}
audit, err := e.sm.TransitionReopenForRevision(doc, actorID, actorName, reason)
if err != nil {
return nil, err
}
e.memAudits[docID] = append(e.memAudits[docID], audit)
if err := e.syncDocAndAuditToDB(ctx, doc, audit); err != nil {
return nil, err
}
return doc, nil
}
// GetDocument 获取文档实体 (优先查内存,未命中查 DB)
func (e *GovernanceEngine) GetDocument(ctx context.Context, docID string) (*DocumentRecord, error) {
e.mu.Lock()
+24
View File
@@ -0,0 +1,24 @@
package docgov
import (
"errors"
"testing"
)
func TestTransitionReopenForRevision_IsNotARejection(t *testing.T) {
sm := NewStateMachine()
doc := &DocumentRecord{DocID: "doc_reopen", CurrentState: StateApproved, IsLocked: true, Config: DefaultDocumentConfig()}
audit, err := sm.TransitionReopenForRevision(doc, 4, "creator", "新材料")
if err != nil {
t.Fatal(err)
}
if doc.CurrentState != StateDrafting || doc.IsLocked {
t.Fatalf("state %s locked %v", doc.CurrentState, doc.IsLocked)
}
if audit.Action != "reopen_for_revision" || audit.ActorRole != RoleSystem {
t.Fatalf("audit %+v", audit)
}
if _, err := sm.TransitionReopenForRevision(doc, 4, "creator", "again"); !errors.Is(err, ErrInvalidStateTransition) {
t.Fatalf("drafting reopen err %v", err)
}
}
+24
View File
@@ -283,3 +283,27 @@ func (sm *StateMachine) TransitionApproverReject(doc *DocumentRecord, actorID in
CreatedAt: now,
}, nil
}
// TransitionReopenForRevision unlocks an approved document for another draft.
// It is not a rejection: the previous approval stays in the audit as a reopen.
func (sm *StateMachine) TransitionReopenForRevision(doc *DocumentRecord, actorID int64, actorName, reason string) (*AuditLog, error) {
if doc == nil {
return nil, ErrDocumentNotFound
}
if doc.CurrentState != StateApproved {
return nil, fmt.Errorf("%w: 只有已批准文档可以再打开编制,当前为【%s】", ErrInvalidStateTransition, doc.CurrentState)
}
if strings.TrimSpace(reason) == "" {
return nil, fmt.Errorf("%w: 再打开必须填写原因", ErrInvalidStateTransition)
}
from := doc.CurrentState
doc.CurrentState = StateDrafting
doc.IsLocked = false
now := time.Now()
doc.UpdatedAt = now
return &AuditLog{
DocID: doc.DocID, Action: "reopen_for_revision", ActorID: actorID, ActorName: actorName,
ActorRole: RoleSystem, FromState: from, ToState: StateDrafting,
Details: map[string]any{"reason": reason, "timestamp": now}, CreatedAt: now,
}, nil
}
+2
View File
@@ -23,6 +23,8 @@ var (
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")
ErrStageNotCompleted = errors.New("stage is not completed")
ErrRecheckAckRequired = errors.New("recheck acknowledgement required")
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")
+13
View File
@@ -28,6 +28,19 @@ type Service struct {
authorOnlyPolicy AuthorOnlyPolicySource
projector DocumentDutyProjector
busyChecker DocTaskBusyChecker
recheck RecheckGate
}
// RecheckGate is the soft hold on a completed stage. Acknowledgement is not clearance.
type RecheckGate interface {
StageNeedsAck(ctx context.Context, projectID int64, stageKey string) (bool, error)
Acknowledge(ctx context.Context, projectID int64, stageKey string, userID int64) error
}
func (s *Service) SetRecheckGate(g RecheckGate) {
if s != nil {
s.recheck = g
}
}
func NewService(store Store, wfStore VersionResolver, projects projectworkflow.Store, docs DocumentReadinessChecker) *Service {
@@ -0,0 +1,139 @@
package projectinstance
import (
"context"
"errors"
"testing"
)
type ackGate struct {
need bool
acked int
}
func (g *ackGate) StageNeedsAck(context.Context, int64, string) (bool, error) {
return g.need, nil
}
func (g *ackGate) Acknowledge(context.Context, int64, string, int64) error {
g.acked++
return nil
}
func TestReopen_KeepsLaterStageActive(t *testing.T) {
ctx := context.Background()
store := newMemStore()
_ = SeedProjectMeta(store, 42, LifecyclePlanning, "", 1, 0)
svc := NewService(store, testResolverMultiStage(), nil, &stubDocChecker{states: map[string]string{
"intake": "approved", "feasibility_reg": "approved",
}})
plan := validPlan()
plan.Stages = append(plan.Stages, StagePlan{
StageKey: "V0.5_feasibility", Enabled: true,
DocTasks: []DocTaskPlan{{
TaskKey: "feasibility_reg", Enabled: true,
Duties: DocDuty{AuthorUserID: 1, ReviewerUserID: 2, ApproverUserID: 3},
}},
})
if _, err := svc.SaveWorkflowPlan(ctx, 42, 1, true, SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1}); err != nil {
t.Fatalf("save: %v", err)
}
preview, err := svc.ActivationPreview(ctx, 42, 1, true)
if err != nil {
t.Fatalf("preview: %v", err)
}
activated, err := svc.Activate(ctx, 42, 1, true, ActivateInput{
ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion,
ExpectedPlanRevision: preview.ExpectedPlanRevision,
})
if err != nil {
t.Fatalf("activate: %v", err)
}
advanced, err := svc.AdvanceStage(ctx, activated.Instance.ID, 1, true, AdvanceStageInput{
StageKey: "V0.1_triage", ExpectedConfigRevision: activated.Instance.ConfigRevision,
})
if err != nil {
t.Fatalf("advance: %v", err)
}
if err := svc.ReopenStage(ctx, activated.Instance.ID, "V0.1_triage", []string{"intake"}, 1); err != nil {
t.Fatalf("reopen: %v", err)
}
reopened, err := svc.GetInstanceDetail(ctx, activated.Instance.ID)
if err != nil {
t.Fatal(err)
}
var triage, feasibility string
for _, st := range reopened.Stages {
switch st.StageKey {
case "V0.1_triage":
triage = st.Status
case "V0.5_feasibility":
feasibility = st.Status
}
}
if triage != StageActive || feasibility != StageActive {
t.Fatalf("triage %s feasibility %s", triage, feasibility)
}
again, err := svc.AdvanceStage(ctx, activated.Instance.ID, 1, true, AdvanceStageInput{
StageKey: "V0.1_triage", ExpectedConfigRevision: reopened.Instance.ConfigRevision,
})
if err != nil {
t.Fatalf("second advance: %v", err)
}
if again.NextStageKey != "" && again.NextStageKey != "V0.5_feasibility" {
t.Fatalf("unexpected next %s", again.NextStageKey)
}
detail, err := svc.GetInstanceDetail(ctx, activated.Instance.ID)
if err != nil {
t.Fatal(err)
}
for _, st := range detail.Stages {
if st.StageKey == "V0.5_feasibility" && st.Status != StageActive {
t.Fatalf("downstream became %s", st.Status)
}
if st.StageKey == "V0.1_triage" && st.Status != StageCompleted {
t.Fatalf("reopened stage became %s", st.Status)
}
}
_ = advanced
}
func TestAdvance_RecheckAckDoesNotClearByItself(t *testing.T) {
ctx := context.Background()
store := newMemStore()
_ = SeedProjectMeta(store, 9, LifecyclePlanning, "", 1, 0)
svc := NewService(store, testResolver(), nil, &stubDocChecker{states: map[string]string{"intake": "approved"}})
gate := &ackGate{need: true}
svc.SetRecheckGate(gate)
if _, err := svc.SaveWorkflowPlan(ctx, 9, 1, true, SavePlanInput{Plan: validPlan(), ExpectedPlanRevision: 0}); err != nil {
t.Fatal(err)
}
preview, err := svc.ActivationPreview(ctx, 9, 1, true)
if err != nil {
t.Fatal(err)
}
result, err := svc.Activate(ctx, 9, 1, true, ActivateInput{
ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion,
ExpectedPlanRevision: preview.ExpectedPlanRevision,
})
if err != nil {
t.Fatal(err)
}
_, err = svc.AdvanceStage(ctx, result.Instance.ID, 1, true, AdvanceStageInput{
StageKey: "V0.1_triage", ExpectedConfigRevision: result.Instance.ConfigRevision,
})
if !errors.Is(err, ErrRecheckAckRequired) {
t.Fatalf("expected ack required, got %v", err)
}
if gate.acked != 0 {
t.Fatal("ack recorded before confirm")
}
if _, err := svc.AdvanceStage(ctx, result.Instance.ID, 1, true, AdvanceStageInput{
StageKey: "V0.1_triage", ExpectedConfigRevision: result.Instance.ConfigRevision, AcknowledgeRecheck: true,
}); err != nil {
t.Fatal(err)
}
if gate.acked != 1 {
t.Fatalf("acked %d", gate.acked)
}
}
+43 -9
View File
@@ -50,16 +50,31 @@ func (s *Service) GetStageReadiness(ctx context.Context, instanceID int64, stage
if stageDef != nil && stageDef.Name != "" {
stageName = stageDef.Name
}
stageIsActive := false
for _, st := range detail.Stages {
if st.StageKey == stageKey && st.Status == StageActive {
stageIsActive = true
break
}
}
needsAck := false
if s.recheck != nil {
needsAck, err = s.recheck.StageNeedsAck(ctx, detail.Instance.ProjectID, stageKey)
if err != nil {
return nil, err
}
}
return &StageReadiness{
InstanceID: instanceID,
StageKey: stageKey,
StageName: stageName,
Ready: ready,
CanAdvance: isCreator && ready && detail.ActiveStageKey == stageKey,
BlockingIssues: issues,
DocTasks: tasks,
ConfigRevision: detail.Instance.ConfigRevision,
InstanceID: instanceID,
StageKey: stageKey,
StageName: stageName,
Ready: ready,
CanAdvance: isCreator && ready && stageIsActive,
RequiresRecheckAck: needsAck,
BlockingIssues: issues,
DocTasks: tasks,
ConfigRevision: detail.Instance.ConfigRevision,
}, nil
}
@@ -80,11 +95,18 @@ func (s *Service) AdvanceStage(ctx context.Context, instanceID int64, userID int
if in.ExpectedConfigRevision != readiness.ConfigRevision {
return nil, ErrInstanceRevisionConflict
}
before, err := s.store.GetInstanceByID(ctx, instanceID)
if err != nil {
return nil, err
}
if readiness.RequiresRecheckAck && !in.AcknowledgeRecheck {
return nil, ErrRecheckAckRequired
}
if readiness.RequiresRecheckAck && s.recheck != nil {
if err := s.recheck.Acknowledge(ctx, before.Instance.ProjectID, in.StageKey, userID); err != nil {
return nil, err
}
}
completedStage := before.ActiveStageKey
var completedStageInstanceID int64
for _, st := range before.Stages {
@@ -108,3 +130,15 @@ func (s *Service) AdvanceStage(ctx context.Context, instanceID int64, userID int
InstanceDone: after.Instance.Status == InstanceCompleted,
}, nil
}
// ReopenStage opens a completed stage again. Other stages stay as they are.
func (s *Service) ReopenStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) error {
if !s.Available() {
return ErrNotAvailable
}
if stageKey == "" || len(docTaskKeys) == 0 {
return ErrInvalidStage
}
_, err := s.store.ReopenStage(ctx, instanceID, stageKey, docTaskKeys, operatorID)
return err
}
+98 -6
View File
@@ -118,17 +118,35 @@ func (s *pgStore) AdvanceStage(ctx context.Context, instanceID int64, stageKey s
if err != nil {
return nil, err
}
var otherActive int
if err = tx.QueryRow(ctx, `
SELECT COUNT(*) FROM project_stage_instances
WHERE workflow_instance_id = $1 AND enabled = true AND status = 'active'
`, instanceID).Scan(&otherActive); 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'
ORDER BY position ASC, id ASC LIMIT 1
`, instanceID).Scan(&nextStageID, &nextStageKey)
if errors.Is(err, pgx.ErrNoRows) {
err = pgx.ErrNoRows
if otherActive == 0 {
err = tx.QueryRow(ctx, `
SELECT id, stage_key FROM project_stage_instances
WHERE workflow_instance_id = $1 AND enabled = true AND status = 'pending'
ORDER BY position ASC, id ASC LIMIT 1
`, instanceID).Scan(&nextStageID, &nextStageKey)
}
if otherActive > 0 {
_, err = tx.Exec(ctx, `
UPDATE project_workflow_instances SET status = 'active', config_revision = config_revision + 1, updated_at = $2
WHERE id = $1
`, instanceID, now)
if err != nil {
return nil, err
}
} else if errors.Is(err, pgx.ErrNoRows) {
_, err = tx.Exec(ctx, `
UPDATE project_workflow_instances SET status = 'completed', config_revision = config_revision + 1,
completed_at = $2, completed_by = $3, updated_at = $2
@@ -201,6 +219,80 @@ func (s *pgStore) MarkDocTaskApproved(ctx context.Context, projectID int64, task
return err
}
func (s *pgStore) ReopenStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) (*InstanceDetail, error) {
tx, err := s.pool.Begin(ctx)
if err != nil {
return nil, err
}
defer tx.Rollback(ctx)
var instStatus string
err = tx.QueryRow(ctx, `
SELECT status FROM project_workflow_instances WHERE id = $1 FOR UPDATE
`, instanceID).Scan(&instStatus)
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrInstanceNotFound
}
if err != nil {
return nil, err
}
if instStatus != InstanceActive && instStatus != InstanceCompleted {
return nil, fmt.Errorf("%w: instance status %s", ErrInvalidStage, instStatus)
}
var stageID int64
var stageStatus string
err = tx.QueryRow(ctx, `
SELECT id, status FROM project_stage_instances
WHERE workflow_instance_id = $1 AND stage_key = $2 AND enabled = true
`, instanceID, stageKey).Scan(&stageID, &stageStatus)
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrInvalidStage
}
if err != nil {
return nil, err
}
if stageStatus != StageCompleted {
return nil, ErrStageNotCompleted
}
now := time.Now()
if _, err = tx.Exec(ctx, `
UPDATE project_stage_instances SET status = 'active', updated_at = $2 WHERE id = $1
`, stageID, now); err != nil {
return nil, err
}
if len(docTaskKeys) > 0 {
if _, err = tx.Exec(ctx, `
UPDATE project_doc_task_instances
SET status = 'in_progress', updated_at = $3
WHERE stage_instance_id = $1 AND task_key = ANY($2) AND status = 'approved'
`, stageID, docTaskKeys, now); err != nil {
return nil, err
}
}
if _, err = tx.Exec(ctx, `
UPDATE project_workflow_instances
SET status = 'active', config_revision = config_revision + 1, updated_at = $2,
completed_at = NULL, completed_by = NULL
WHERE id = $1
`, instanceID, now); err != nil {
return nil, err
}
_ = operatorID
if err := tx.Commit(ctx); err != nil {
return nil, err
}
row := s.pool.QueryRow(ctx, `
SELECT 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
FROM project_workflow_instances WHERE id = $1
`, instanceID)
inst, err := scanInstance(row)
if err != nil {
return nil, err
}
return s.loadInstanceDetail(ctx, inst)
}
type dbExec interface {
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
}
+65 -4
View File
@@ -57,6 +57,8 @@ type Store interface {
GetInstanceByID(ctx context.Context, instanceID int64) (*InstanceDetail, error)
AdvanceStage(ctx context.Context, instanceID int64, stageKey string, expectedRevision int, operatorID int64) (*InstanceDetail, error)
// ReopenStage sets a completed stage back to active without changing other stages.
ReopenStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) (*InstanceDetail, error)
MarkDocTaskApproved(ctx context.Context, projectID int64, taskKey string) error
// UpdateLiveDocTaskDuties 更新 live 实例 config_snapshot 与 document_duty_assignments(单任务)。
UpdateLiveDocTaskDuties(ctx context.Context, projectID int64, taskKey string, plan WorkflowPlan, assignedBy int64) (*WorkflowInstance, error)
@@ -1148,16 +1150,27 @@ func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey st
return nil, ErrInvalidStage
}
records[activeIdx].status = StageCompleted
nextIdx := -1
anotherActive := false
for i, rec := range records {
if rec.enabled && rec.status == StagePending {
nextIdx = i
if i != activeIdx && rec.enabled && rec.status == StageActive {
anotherActive = true
break
}
}
nextIdx := -1
if !anotherActive {
for i, rec := range records {
if rec.enabled && rec.status == StagePending {
nextIdx = i
break
}
}
}
inst.ConfigRevision++
inst.UpdatedAt = time.Now()
if nextIdx < 0 {
if anotherActive {
inst.Status = InstanceActive
} else if nextIdx < 0 {
inst.Status = InstanceCompleted
if pm := m.meta[inst.ProjectID]; pm.lifecycle != "" {
pm.lifecycle = LifecyclePlanning
@@ -1173,6 +1186,54 @@ func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey st
return m.instanceDetailLocked(&copyInst), nil
}
func (m *memStore) ReopenStage(_ context.Context, instanceID int64, stageKey string, docTaskKeys []string, _ int64) (*InstanceDetail, error) {
m.mu.Lock()
defer m.mu.Unlock()
var inst *WorkflowInstance
for _, live := range m.live {
if live.ID == instanceID {
inst = live
break
}
}
if inst == nil {
return nil, ErrInstanceNotFound
}
records := m.stages[instanceID]
idx := -1
for i, rec := range records {
if rec.stageKey == stageKey {
idx = i
break
}
}
if idx < 0 {
return nil, ErrInvalidStage
}
if records[idx].status != StageCompleted {
return nil, ErrStageNotCompleted
}
records[idx].status = StageActive
wanted := map[string]bool{}
for _, key := range docTaskKeys {
wanted[key] = true
}
for j := range records[idx].tasks {
if wanted[records[idx].tasks[j].taskKey] {
records[idx].tasks[j].status = "in_progress"
}
}
if inst.Status == InstanceCompleted {
inst.Status = InstanceActive
}
inst.ConfigRevision++
inst.UpdatedAt = time.Now()
m.stages[instanceID] = records
copyInst := *inst
m.live[inst.ProjectID] = &copyInst
return m.instanceDetailLocked(&copyInst), nil
}
func (m *memStore) MarkDocTaskApproved(_ context.Context, projectID int64, taskKey string) error {
m.mu.Lock()
defer m.mu.Unlock()
+6 -2
View File
@@ -240,8 +240,9 @@ 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"`
Ready bool `json:"ready"`
CanAdvance bool `json:"can_advance"`
RequiresRecheckAck bool `json:"requires_recheck_ack"`
BlockingIssues []GateIssue `json:"blocking_issues"`
DocTasks []DocTaskReadiness `json:"doc_tasks"`
ConfigRevision int `json:"config_revision"`
@@ -251,6 +252,9 @@ type StageReadiness struct {
type AdvanceStageInput struct {
StageKey string `json:"stage_key"`
ExpectedConfigRevision int `json:"expected_config_revision"`
// AcknowledgeRecheck records that the creator saw open recheck marks.
// It does not clear those marks.
AcknowledgeRecheck bool `json:"acknowledge_recheck"`
}
// AdvanceStageResult 阶段推进结果。