refactor(projectinstance): split store.go into use-case files

Move mem/pg activate/live/meta/plan/token/seed paths into ≤800-line
siblings without behavior change; short tests green.

internal/projectinstance/store_mem.go

@@ -0,0 +1,579 @@
+package projectinstance
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"chat-agent-system/internal/workflowdef"
+)
+
+// =============================================================================
+// 内存 Store（单元测试 / 无 PG 迁移）
+// =============================================================================
+
+type memStageRecord struct {
+	id       int64
+	stageKey string
+	position int
+	enabled  bool
+	status   string
+	tasks    []memDocTaskRecord
+}
+
+type memDocTaskRecord struct {
+	id      int64
+	taskKey string
+	enabled bool
+	status  string
+}
+
+type memStore struct {
+	mu          sync.Mutex
+	meta        map[int64]projectMeta
+	plan        map[int64]*WorkflowInstance
+	live        map[int64]*WorkflowInstance
+	tokens      map[string]activationTokenRec
+	stages      map[int64][]memStageRecord
+	nextStageID int64
+	nextTaskID  int64
+}
+
+type projectMeta struct {
+	lifecycle    string
+	activeKey    string
+	selectionVer int
+	planRevision int
+}
+
+type activationTokenRec struct {
+	projectID      int64
+	planHash       string
+	expectedVer    int
+	snapshot       WorkflowPlan
+	expiresAt      time.Time
+	used           bool
+	policyRevision *int64
+}
+
+func newMemStore() *memStore {
+	return &memStore{
+		meta:        make(map[int64]projectMeta),
+		plan:        make(map[int64]*WorkflowInstance),
+		live:        make(map[int64]*WorkflowInstance),
+		tokens:      make(map[string]activationTokenRec),
+		stages:      make(map[int64][]memStageRecord),
+		nextStageID: 1,
+		nextTaskID:  1,
+	}
+}
+
+func (m *memStore) rebuildStagesLocked(inst *WorkflowInstance, activate bool) {
+	records := make([]memStageRecord, 0, len(inst.ConfigSnapshot.Stages))
+	firstActive := true
+	for i, st := range inst.ConfigSnapshot.Stages {
+		rec := memStageRecord{
+			id: m.nextStageID, stageKey: st.StageKey, position: i, enabled: st.Enabled,
+		}
+		m.nextStageID++
+		if !st.Enabled {
+			rec.status = StageSkipped
+		} else if activate && firstActive {
+			rec.status = StageActive
+			firstActive = false
+		} else {
+			rec.status = StagePending
+		}
+		for _, dt := range st.DocTasks {
+			task := memDocTaskRecord{id: m.nextTaskID, taskKey: dt.TaskKey, enabled: dt.Enabled, status: "pending"}
+			m.nextTaskID++
+			if !dt.Enabled {
+				task.status = "skipped_disabled"
+			}
+			rec.tasks = append(rec.tasks, task)
+		}
+		records = append(records, rec)
+	}
+	m.stages[inst.ID] = records
+}
+
+func (m *memStore) instanceDetailLocked(inst *WorkflowInstance) *InstanceDetail {
+	detail := &InstanceDetail{Instance: *inst, Stages: make([]StageInstance, 0)}
+	for _, rec := range m.stages[inst.ID] {
+		st := StageInstance{
+			ID: rec.id, StageKey: rec.stageKey, Position: rec.position, Enabled: rec.enabled, Status: rec.status,
+		}
+		if st.Status == StageActive {
+			detail.ActiveStageKey = st.StageKey
+		}
+		for _, t := range rec.tasks {
+			st.DocTasks = append(st.DocTasks, DocTaskInstance{
+				ID: t.id, TaskKey: t.taskKey, Enabled: t.enabled, Status: t.status, Approved: t.status == "approved",
+			})
+		}
+		detail.Stages = append(detail.Stages, st)
+	}
+	return detail
+}
+
+func (m *memStore) ensureMeta(projectID int64) projectMeta {
+	if pm, ok := m.meta[projectID]; ok {
+		return pm
+	}
+	pm := projectMeta{lifecycle: LifecyclePlanning, selectionVer: 1}
+	m.meta[projectID] = pm
+	return pm
+}
+
+func (m *memStore) GetProjectMeta(_ context.Context, projectID int64) (string, string, int, int, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	pm, ok := m.meta[projectID]
+	if !ok {
+		return "", "", 0, 0, ErrProjectNotFound
+	}
+	return pm.lifecycle, pm.activeKey, pm.selectionVer, pm.planRevision, nil
+}
+
+func (m *memStore) UpdateProjectAfterActivate(_ context.Context, projectID int64, workflowKey string, expectedVersion int, _ int64) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	pm := m.ensureMeta(projectID)
+	if pm.selectionVer != expectedVersion {
+		return ErrConcurrentConflict
+	}
+	pm.lifecycle = LifecycleActive
+	pm.activeKey = workflowKey
+	pm.selectionVer++
+	m.meta[projectID] = pm
+	return nil
+}
+
+func (m *memStore) BumpPlanRevision(_ context.Context, projectID int64, expectedPlanRevision int) (int, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	pm := m.ensureMeta(projectID)
+	if pm.planRevision != expectedPlanRevision {
+		return 0, ErrPlanRevisionConflict
+	}
+	pm.planRevision++
+	m.meta[projectID] = pm
+	return pm.planRevision, nil
+}
+
+func (m *memStore) GetPlanningInstance(_ context.Context, projectID int64) (*WorkflowInstance, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	inst := m.plan[projectID]
+	if inst == nil {
+		return nil, nil
+	}
+	copyInst := *inst
+	return &copyInst, nil
+}
+
+func (m *memStore) GetLiveInstance(_ context.Context, projectID int64) (*WorkflowInstance, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	inst := m.live[projectID]
+	if inst == nil || !IsLiveInstanceStatus(inst.Status) {
+		return nil, nil
+	}
+	copyInst := *inst
+	return &copyInst, nil
+}
+
+func (m *memStore) GetGoverningInstanceForAccess(_ context.Context, projectID int64) (*WorkflowInstance, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	inst := m.live[projectID]
+	if inst == nil {
+		return nil, nil
+	}
+	if IsLiveInstanceStatus(inst.Status) || IsClosedInstanceStatus(inst.Status) {
+		copyInst := *inst
+		return &copyInst, nil
+	}
+	// planning 等非治理态：视为从未激活
+	return nil, nil
+}
+
+func (m *memStore) HasLiveInstance(_ context.Context, projectID int64) (bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	inst := m.live[projectID]
+	return inst != nil && IsLiveInstanceStatus(inst.Status), nil
+}
+
+func (m *memStore) UpsertPlanningInstance(_ context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if inst.ID == 0 {
+		inst.ID = int64(len(m.plan) + 1)
+	}
+	inst.Status = InstancePlanning
+	now := time.Now()
+	if inst.CreatedAt.IsZero() {
+		inst.CreatedAt = now
+	}
+	inst.UpdatedAt = now
+	copyInst := *inst
+	m.plan[inst.ProjectID] = &copyInst
+	m.rebuildStagesLocked(&copyInst, false)
+	return &copyInst, nil
+}
+
+// SavePlanWithRevision 内存实现：bump 与 upsert 在同一互斥锁内，天然原子。
+func (m *memStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) {
+	newRev, err := m.BumpPlanRevision(ctx, inst.ProjectID, expectedPlanRevision)
+	if err != nil {
+		return nil, 0, err
+	}
+	saved, err := m.UpsertPlanningInstance(ctx, inst)
+	if err != nil {
+		return nil, 0, err
+	}
+	return saved, newRev, nil
+}
+
+func (m *memStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
+	usesAuthorOnly := planUsesAuthorOnly(inst.ConfigSnapshot)
+	var currentPolicyRev int64
+	if usesAuthorOnly {
+		snap, lockErr, ok := runActivateAuthorOnlyLock(ctx, nil)
+		if !ok {
+			return nil, ErrAuthorOnlyPolicyUnavailable
+		}
+		if lockErr != nil {
+			return nil, fmt.Errorf("%w: %v", ErrAuthorOnlyPolicyUnavailable, lockErr)
+		}
+		if !snap.Allowed {
+			return nil, ErrAuthorOnlyPolicyDisabled
+		}
+		currentPolicyRev = snap.Revision
+	}
+	if err := runActivateMemberRecheck(ctx); err != nil {
+		return nil, err
+	}
+	tokenPolicyRev, err := m.consumeActivationTokenLocked(ctx, token, projectID, planHash)
+	if err != nil {
+		return nil, err
+	}
+	if usesAuthorOnly {
+		if tokenPolicyRev == nil || *tokenPolicyRev != currentPolicyRev {
+			return nil, ErrAuthorOnlyPolicyChanged
+		}
+	}
+	return m.ActivateInstance(ctx, projectID, inst, version, expectedVersion, activatedBy)
+}
+
+func (m *memStore) CountActiveTaskRuns(_ context.Context, _ int64) (int, error) {
+	return 0, nil
+}
+
+func (m *memStore) ActivateInstance(_ context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
+	workflowKey := ""
+	if version != nil {
+		workflowKey = version.WorkflowKey
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	pm := m.ensureMeta(projectID)
+	if pm.selectionVer != expectedVersion {
+		return nil, ErrConcurrentConflict
+	}
+	pm.lifecycle = LifecycleActive
+	pm.activeKey = workflowKey
+	pm.selectionVer++
+	m.meta[projectID] = pm
+
+	now := time.Now()
+	activated := *inst
+	activated.Status = InstanceActive
+	activated.ActivatedBy = &activatedBy
+	activated.ActivatedAt = &now
+	m.live[projectID] = &activated
+	m.rebuildStagesLocked(&activated, true)
+	delete(m.plan, projectID)
+	return &activated, nil
+}
+
+func (m *memStore) GetInstanceByID(_ context.Context, instanceID int64) (*InstanceDetail, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, inst := range m.live {
+		if inst.ID == instanceID {
+			return m.instanceDetailLocked(inst), nil
+		}
+	}
+	for _, inst := range m.plan {
+		if inst.ID == instanceID {
+			return m.instanceDetailLocked(inst), nil
+		}
+	}
+	return nil, ErrInstanceNotFound
+}
+
+func (m *memStore) AdvanceStage(_ context.Context, instanceID int64, stageKey string, expectedRevision int, _ 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
+	}
+	if inst.ConfigRevision != expectedRevision {
+		return nil, ErrInstanceRevisionConflict
+	}
+	records := m.stages[instanceID]
+	activeIdx := -1
+	for i, rec := range records {
+		if rec.stageKey == stageKey {
+			if rec.status != StageActive {
+				return nil, ErrStageNotActive
+			}
+			activeIdx = i
+			break
+		}
+	}
+	if activeIdx < 0 {
+		return nil, ErrInvalidStage
+	}
+	records[activeIdx].status = StageCompleted
+	anotherActive := false
+	for i, rec := range records {
+		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 anotherActive {
+		inst.Status = InstanceActive
+	} else 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
+	}
+	m.stages[instanceID] = records
+	copyInst := *inst
+	m.live[inst.ProjectID] = &copyInst
+	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()
+	live := m.live[projectID]
+	if live == nil {
+		return nil
+	}
+	records := m.stages[live.ID]
+	for i := range records {
+		for j := range records[i].tasks {
+			if records[i].tasks[j].taskKey == taskKey && records[i].tasks[j].enabled {
+				records[i].tasks[j].status = "approved"
+			}
+		}
+	}
+	m.stages[live.ID] = records
+	return nil
+}
+
+func (m *memStore) UpdateLiveDocTaskDuties(_ context.Context, projectID int64, taskKey string, plan WorkflowPlan, _ int64) (*WorkflowInstance, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	live := m.live[projectID]
+	if live == nil {
+		return nil, ErrInstanceNotFound
+	}
+	if findDocTaskPlan(plan, taskKey) == nil {
+		return nil, ErrDocTaskNotFound
+	}
+	planHash, err := computePlanHash(plan)
+	if err != nil {
+		return nil, err
+	}
+	copyInst := *live
+	copyInst.ConfigSnapshot = plan
+	copyInst.PlanHash = planHash
+	copyInst.ConfigRevision++
+	copyInst.UpdatedAt = time.Now()
+	m.live[projectID] = &copyInst
+	return &copyInst, nil
+}
+
+func (m *memStore) UpdateLivePlan(_ context.Context, projectID int64, plan WorkflowPlan, expectedConfigRevision int, _ int64) (*WorkflowInstance, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	live := m.live[projectID]
+	if live == nil {
+		return nil, ErrInstanceNotFound
+	}
+	if live.ConfigRevision != expectedConfigRevision {
+		return nil, ErrInstanceRevisionConflict
+	}
+	planHash, err := computePlanHash(plan)
+	if err != nil {
+		return nil, err
+	}
+	copyInst := *live
+	copyInst.ConfigSnapshot = plan
+	copyInst.PlanHash = planHash
+	copyInst.ConfigRevision++
+	copyInst.UpdatedAt = time.Now()
+	m.live[projectID] = &copyInst
+
+	records := m.stages[copyInst.ID]
+	for i := range records {
+		for _, st := range plan.Stages {
+			if records[i].stageKey == st.StageKey {
+				records[i].enabled = st.Enabled
+			}
+		}
+		for j := range records[i].tasks {
+			dt := findDocTaskPlan(plan, records[i].tasks[j].taskKey)
+			if dt == nil {
+				continue
+			}
+			records[i].tasks[j].enabled = dt.Enabled
+			if !dt.Enabled && records[i].tasks[j].status == "pending" {
+				records[i].tasks[j].status = "skipped_disabled"
+			}
+			if dt.Enabled && records[i].tasks[j].status == "skipped_disabled" {
+				records[i].tasks[j].status = "pending"
+			}
+		}
+	}
+	m.stages[copyInst.ID] = records
+	return &copyInst, 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 = InstanceTerminated
+	copyInst := *inst
+	m.live[projectID] = &copyInst
+	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, policyRevision *int64) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var revCopy *int64
+	if policyRevision != nil {
+		v := *policyRevision
+		revCopy = &v
+	}
+	m.tokens[hashActivationToken(token)] = activationTokenRec{
+		projectID: projectID, planHash: planHash, expectedVer: expectedVersion,
+		snapshot: snapshot, expiresAt: expiresAt, policyRevision: revCopy,
+	}
+	return nil
+}
+
+func (m *memStore) ConsumeActivationToken(ctx context.Context, token string, projectID int64, planHash string) (*WorkflowPlan, error) {
+	_, plan, err := m.consumeActivationTokenWithRev(ctx, token, projectID, planHash)
+	return plan, err
+}
+
+func (m *memStore) consumeActivationTokenLocked(ctx context.Context, token string, projectID int64, planHash string) (*int64, error) {
+	rev, _, err := m.consumeActivationTokenWithRev(ctx, token, projectID, planHash)
+	return rev, err
+}
+
+func (m *memStore) consumeActivationTokenWithRev(_ context.Context, token string, projectID int64, planHash string) (*int64, *WorkflowPlan, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	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, nil, ErrTokenInvalid
+	}
+	rec.used = true
+	m.tokens[tokenHash] = rec
+	plan := rec.snapshot
+	return rec.policyRevision, &plan, nil
+}

internal/projectinstance/store_pg_activate.go

@@ -0,0 +1,233 @@
+package projectinstance
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	"chat-agent-system/internal/workflowdef"
+)
+
+func (s *pgStore) ActivateInstance(ctx context.Context, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64) (*WorkflowInstance, error) {
+	return s.activateInstanceTx(ctx, s.pool, "", projectID, inst, version, expectedVersion, activatedBy, "")
+}
+
+func (s *pgStore) ActivateInstanceWithToken(ctx context.Context, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
+	return s.activateInstanceTx(ctx, s.pool, token, projectID, inst, version, expectedVersion, activatedBy, planHash)
+}
+
+func (s *pgStore) activateInstanceTx(ctx context.Context, db pgxDB, token string, projectID int64, inst *WorkflowInstance, version *workflowdef.Version, expectedVersion int, activatedBy int64, planHash string) (*WorkflowInstance, error) {
+	workflowKey := ""
+	if version != nil {
+		workflowKey = version.WorkflowKey
+	}
+	tx, err := db.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback(ctx)
+
+	usesAuthorOnly := planUsesAuthorOnly(inst.ConfigSnapshot)
+	var currentPolicyRev int64
+	if usesAuthorOnly {
+		snap, lockErr, ok := runActivateAuthorOnlyLock(ctx, tx)
+		if !ok {
+			return nil, ErrAuthorOnlyPolicyUnavailable
+		}
+		if lockErr != nil {
+			return nil, fmt.Errorf("%w: %v", ErrAuthorOnlyPolicyUnavailable, lockErr)
+		}
+		if !snap.Allowed {
+			return nil, ErrAuthorOnlyPolicyDisabled
+		}
+		currentPolicyRev = snap.Revision
+	}
+
+	if err := runActivateMemberRecheck(ctx); err != nil {
+		return nil, err
+	}
+	if token != "" {
+		tokenPolicyRev, err := consumeActivationTokenTx(ctx, tx, token, projectID, planHash)
+		if err != nil {
+			return nil, err
+		}
+		if usesAuthorOnly {
+			if tokenPolicyRev == nil || *tokenPolicyRev != currentPolicyRev {
+				return nil, ErrAuthorOnlyPolicyChanged
+			}
+		}
+	}
+
+	tag, err := tx.Exec(ctx, `
+		UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2,
+		       workflow_selection_version = workflow_selection_version + 1,
+		       active_workflow_changed_at = now(), active_workflow_changed_by = $4,
+		       updated_at = now()
+		WHERE id = $1 AND workflow_selection_version = $3
+	`, projectID, workflowKey, expectedVersion, activatedBy)
+	if err != nil {
+		return nil, err
+	}
+	if tag.RowsAffected() == 0 {
+		return nil, ErrConcurrentConflict
+	}
+
+	now := time.Now()
+	row := tx.QueryRow(ctx, `
+		UPDATE project_workflow_instances SET status = 'active', activated_by = $2, activated_at = $3, updated_at = $3
+		WHERE id = $1 AND status = 'planning'
+		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
+	`, inst.ID, activatedBy, now)
+	activated, err := scanInstance(row)
+	if err != nil {
+		return nil, err
+	}
+
+	_, err = tx.Exec(ctx, `
+		UPDATE project_stage_instances SET status = CASE WHEN enabled THEN 'active' ELSE 'skipped' END, updated_at = now()
+		WHERE workflow_instance_id = $1 AND stage_key = (
+			SELECT stage_key FROM project_stage_instances
+			WHERE workflow_instance_id = $1 AND enabled = true
+			ORDER BY id ASC LIMIT 1
+		)
+	`, inst.ID)
+	if err != nil {
+		return nil, err
+	}
+	_, err = tx.Exec(ctx, `
+		UPDATE project_stage_instances SET status = 'pending', updated_at = now()
+		WHERE workflow_instance_id = $1 AND status <> 'active' AND enabled = true
+	`, inst.ID)
+	if err != nil {
+		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
+}
+
+type pgxDB interface {
+	Begin(ctx context.Context) (pgx.Tx, error)
+}
+
+func consumeActivationTokenTx(ctx context.Context, tx pgx.Tx, token string, projectID int64, planHash string) (*int64, error) {
+	tokenHash := hashActivationToken(token)
+	var storedHash string
+	var usedAt *time.Time
+	var expiresAt time.Time
+	var policyRev *int64
+	err := tx.QueryRow(ctx, `
+		SELECT plan_hash, used_at, expires_at, author_only_policy_revision
+		FROM workflow_activation_tokens WHERE token = $1 AND project_id = $2
+	`, tokenHash, projectID).Scan(&storedHash, &usedAt, &expiresAt, &policyRev)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrTokenInvalid
+	}
+	if err != nil {
+		return nil, err
+	}
+	if usedAt != nil || time.Now().After(expiresAt) || storedHash != planHash {
+		return nil, ErrTokenInvalid
+	}
+	tag, err := tx.Exec(ctx, `
+		UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL
+	`, tokenHash)
+	if err != nil {
+		return nil, err
+	}
+	if tag.RowsAffected() == 0 {
+		return nil, ErrTokenInvalid
+	}
+	return policyRev, nil
+}
+
+func (s *pgStore) CountActiveTaskRuns(ctx context.Context, projectID int64) (int, error) {
+	if projectID <= 0 {
+		return 0, nil
+	}
+	var count int
+	err := s.pool.QueryRow(ctx, `
+		SELECT COUNT(*) FROM task_runs
+		WHERE project_id = $1 AND status IN ('pending', 'running')
+	`, projectID).Scan(&count)
+	if err != nil {
+		return 0, nil
+	}
+	return count, nil
+}
+
+func (s *pgStore) applyRuntimeMetadata(ctx context.Context, tx pgx.Tx, instanceID int64, def workflowdef.VersionDefinition) error {
+	taskDefs := map[string]workflowdef.DocTaskDef{}
+	for _, st := range def.Stages {
+		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
+}

internal/projectinstance/store_pg_live.go

@@ -0,0 +1,248 @@
+package projectinstance
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	"chat-agent-system/internal/taskrun"
+)
+
+func (s *pgStore) UpdateLiveDocTaskDuties(ctx context.Context, projectID int64, taskKey string, plan WorkflowPlan, assignedBy int64) (*WorkflowInstance, error) {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback(ctx)
+
+	live, err := s.GetLiveInstance(ctx, projectID)
+	if err != nil {
+		return nil, err
+	}
+	if live == nil {
+		return nil, ErrInstanceNotFound
+	}
+
+	planHash, err := computePlanHash(plan)
+	if err != nil {
+		return nil, err
+	}
+	snapRaw, err := json.Marshal(plan)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now()
+	row := tx.QueryRow(ctx, `
+		UPDATE project_workflow_instances SET
+			config_snapshot = $2, plan_hash = $3, config_revision = config_revision + 1, updated_at = $4
+		WHERE id = $1 AND status IN ('active', 'awaiting_stage_confirmation')
+		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
+	`, live.ID, snapRaw, planHash, now)
+	inst, err := scanInstance(row)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrInstanceNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	var docTaskID int64
+	err = tx.QueryRow(ctx, `
+		SELECT dti.id FROM project_doc_task_instances dti
+		JOIN project_stage_instances si ON si.id = dti.stage_instance_id
+		WHERE si.workflow_instance_id = $1 AND dti.task_key = $2
+	`, inst.ID, taskKey).Scan(&docTaskID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrDocTaskNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	if _, err := tx.Exec(ctx, `DELETE FROM document_duty_assignments WHERE doc_task_instance_id = $1`, docTaskID); err != nil {
+		return nil, err
+	}
+
+	dt := findDocTaskPlan(plan, taskKey)
+	if dt == nil {
+		return nil, ErrDocTaskNotFound
+	}
+	mode, effective, _ := ResolveDocTaskDuty(*dt)
+	if mode != DutyModeNone {
+		for role, uid := range map[string]int64{
+			DutyAuthor: effective.AuthorUserID, DutyReviewer: effective.ReviewerUserID, DutyApprover: effective.ApproverUserID,
+		} {
+			if uid <= 0 {
+				continue
+			}
+			if _, err := tx.Exec(ctx, `
+				INSERT INTO document_duty_assignments (doc_task_instance_id, duty_role, user_id, assigned_by)
+				VALUES ($1, $2, $3, $4)
+			`, docTaskID, role, uid, assignedBy); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return nil, err
+	}
+	return inst, nil
+}
+
+func (s *pgStore) UpdateLivePlan(ctx context.Context, projectID int64, plan WorkflowPlan, expectedConfigRevision int, assignedBy int64) (*WorkflowInstance, error) {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback(ctx)
+
+	var instanceID int64
+	var revision int
+	err = tx.QueryRow(ctx, `
+		SELECT id, config_revision
+		FROM project_workflow_instances
+		WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation')
+		FOR UPDATE
+	`, projectID).Scan(&instanceID, &revision)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrInstanceNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	if revision != expectedConfigRevision {
+		return nil, ErrInstanceRevisionConflict
+	}
+
+	planHash, err := computePlanHash(plan)
+	if err != nil {
+		return nil, err
+	}
+	snapRaw, err := json.Marshal(plan)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now()
+	row := tx.QueryRow(ctx, `
+		UPDATE project_workflow_instances SET
+			config_snapshot = $2, plan_hash = $3, config_revision = config_revision + 1, updated_at = $4
+		WHERE id = $1 AND config_revision = $5
+		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, snapRaw, planHash, now, expectedConfigRevision)
+	inst, err := scanInstance(row)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrInstanceRevisionConflict
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	for _, st := range plan.Stages {
+		if _, err := tx.Exec(ctx, `
+			UPDATE project_stage_instances
+			SET enabled = $3, updated_at = $4
+			WHERE workflow_instance_id = $1 AND stage_key = $2
+		`, inst.ID, st.StageKey, st.Enabled, now); err != nil {
+			return nil, err
+		}
+		for _, dt := range st.DocTasks {
+			var docTaskID int64
+			err = tx.QueryRow(ctx, `
+				UPDATE project_doc_task_instances AS dti SET
+					enabled = $3,
+					status = CASE
+						WHEN $3 = false AND dti.status = 'pending' THEN 'skipped_disabled'
+						WHEN $3 = true AND dti.status = 'skipped_disabled' THEN 'pending'
+						ELSE dti.status
+					END,
+					updated_at = $4
+				FROM project_stage_instances si
+				WHERE dti.stage_instance_id = si.id
+				  AND si.workflow_instance_id = $1
+				  AND dti.task_key = $2
+				RETURNING dti.id
+			`, inst.ID, dt.TaskKey, dt.Enabled, now).Scan(&docTaskID)
+			if errors.Is(err, pgx.ErrNoRows) {
+				return nil, ErrDocTaskNotFound
+			}
+			if err != nil {
+				return nil, err
+			}
+			if _, err := tx.Exec(ctx, `DELETE FROM document_duty_assignments WHERE doc_task_instance_id = $1`, docTaskID); err != nil {
+				return nil, err
+			}
+			if !dt.Enabled || !st.Enabled {
+				continue
+			}
+			mode, effective, _ := ResolveDocTaskDuty(dt)
+			if mode == DutyModeNone {
+				continue
+			}
+			for role, uid := range map[string]int64{
+				DutyAuthor: effective.AuthorUserID, DutyReviewer: effective.ReviewerUserID, DutyApprover: effective.ApproverUserID,
+			} {
+				if uid <= 0 {
+					continue
+				}
+				if _, err := tx.Exec(ctx, `
+					INSERT INTO document_duty_assignments (doc_task_instance_id, duty_role, user_id, assigned_by)
+					VALUES ($1, $2, $3, $4)
+				`, docTaskID, role, uid, assignedBy); err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return nil, err
+	}
+	return inst, 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 := taskrun.CancelOpenByProject(ctx, tx, projectID, now, "workflow terminated"); err != nil {
+		return nil, err
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return nil, err
+	}
+	return inst, nil
+}

internal/projectinstance/store_pg_meta.go

@@ -0,0 +1,114 @@
+package projectinstance
+
+import (
+	"context"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+)
+
+func (s *pgStore) GetProjectMeta(ctx context.Context, projectID int64) (string, string, int, int, error) {
+	var lifecycle, activeKey string
+	var selVer, planRev int
+	err := s.pool.QueryRow(ctx, `
+		SELECT COALESCE(lifecycle_status, 'planning'), COALESCE(active_workflow_key, ''),
+		       COALESCE(workflow_selection_version, 1), COALESCE(workflow_plan_revision, 0)
+		FROM projects WHERE id = $1
+	`, projectID).Scan(&lifecycle, &activeKey, &selVer, &planRev)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return "", "", 0, 0, ErrProjectNotFound
+	}
+	return lifecycle, activeKey, selVer, planRev, err
+}
+
+func (s *pgStore) UpdateProjectAfterActivate(ctx context.Context, projectID int64, workflowKey string, expectedVersion int, activatedBy int64) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE projects SET lifecycle_status = 'active', active_workflow_key = $2,
+		       workflow_selection_version = workflow_selection_version + 1,
+		       active_workflow_changed_at = now(), active_workflow_changed_by = $4,
+		       updated_at = now()
+		WHERE id = $1 AND workflow_selection_version = $3
+	`, projectID, workflowKey, expectedVersion, activatedBy)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrConcurrentConflict
+	}
+	return nil
+}
+
+func (s *pgStore) BumpPlanRevision(ctx context.Context, projectID int64, expectedPlanRevision int) (int, error) {
+	var newRev int
+	err := s.pool.QueryRow(ctx, `
+		UPDATE projects SET workflow_plan_revision = workflow_plan_revision + 1, updated_at = now()
+		WHERE id = $1 AND workflow_plan_revision = $2
+		RETURNING workflow_plan_revision
+	`, projectID, expectedPlanRevision).Scan(&newRev)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return 0, ErrPlanRevisionConflict
+	}
+	return newRev, err
+}
+
+func (s *pgStore) GetPlanningInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
+	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 project_id = $1 AND status = 'planning'
+		ORDER BY id DESC LIMIT 1
+	`, projectID)
+	inst, err := scanInstance(row)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	return inst, err
+}
+
+func (s *pgStore) GetLiveInstance(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
+	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 project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
+		ORDER BY id DESC LIMIT 1
+	`, projectID)
+	inst, err := scanInstance(row)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	return inst, err
+}
+
+func (s *pgStore) GetGoverningInstanceForAccess(ctx context.Context, projectID int64) (*WorkflowInstance, error) {
+	live, err := s.GetLiveInstance(ctx, projectID)
+	if err != nil {
+		return nil, err
+	}
+	if live != nil {
+		return live, nil
+	}
+	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 project_id = $1 AND status IN ('completed', 'terminated')
+		ORDER BY id DESC LIMIT 1
+	`, projectID)
+	inst, err := scanInstance(row)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil
+	}
+	return inst, err
+}
+
+func (s *pgStore) HasLiveInstance(ctx context.Context, projectID int64) (bool, error) {
+	var ok bool
+	err := s.pool.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1 FROM project_workflow_instances
+			WHERE project_id = $1 AND status IN ('active', 'awaiting_stage_confirmation', 'termination_requested')
+		)
+	`, projectID).Scan(&ok)
+	return ok, err
+}

internal/projectinstance/store_pg_plan.go

@@ -0,0 +1,227 @@
+package projectinstance
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+func (s *pgStore) UpsertPlanningInstance(ctx context.Context, inst *WorkflowInstance) (*WorkflowInstance, error) {
+	snapRaw, _ := json.Marshal(inst.ConfigSnapshot)
+	now := time.Now()
+	if inst.CreatedAt.IsZero() {
+		inst.CreatedAt = now
+	}
+	inst.UpdatedAt = now
+
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback(ctx)
+
+	var existingID int64
+	err = tx.QueryRow(ctx, `
+		SELECT id FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning' LIMIT 1
+	`, inst.ProjectID).Scan(&existingID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		// Allocate next instance_no so planning drafts after a completed main/rdev
+		// instance do not collide on (project_id, instance_no).
+		var nextNo int
+		if err := tx.QueryRow(ctx, `
+			SELECT COALESCE(MAX(instance_no), 0) + 1
+			FROM project_workflow_instances
+			WHERE project_id = $1
+		`, inst.ProjectID).Scan(&nextNo); err != nil {
+			return nil, err
+		}
+		inst.InstanceNo = nextNo
+		err = tx.QueryRow(ctx, `
+			INSERT INTO project_workflow_instances (
+				project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
+				config_snapshot, created_by, created_at, updated_at
+			) VALUES ($1, $2, $3, 'planning', $4, $5, $6, $7, $8, $9)
+			RETURNING id
+		`, inst.ProjectID, inst.WorkflowVersionID, nextNo, inst.ConfigRevision, inst.PlanHash, snapRaw,
+			inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID)
+		if err != nil {
+			return nil, err
+		}
+	} else if err != nil {
+		return nil, err
+	} else {
+		err = tx.QueryRow(ctx, `
+			UPDATE project_workflow_instances SET
+				workflow_version_id = $2, config_revision = $3, plan_hash = $4, config_snapshot = $5,
+				updated_at = $6
+			WHERE id = $1 AND status = 'planning'
+			RETURNING id
+		`, existingID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw, inst.UpdatedAt).Scan(&inst.ID)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, err
+	}
+	return inst, nil
+}
+
+// SavePlanWithRevision 在单事务内完成 planning 实例 upsert 与 plan revision bump。
+// bump 先执行（带期望版本乐观锁），失败则整个事务回滚，保证 plan 内容与版本号原子一致。
+func (s *pgStore) SavePlanWithRevision(ctx context.Context, inst *WorkflowInstance, expectedPlanRevision int) (*WorkflowInstance, int, error) {
+	snapRaw, _ := json.Marshal(inst.ConfigSnapshot)
+	now := time.Now()
+	if inst.CreatedAt.IsZero() {
+		inst.CreatedAt = now
+	}
+	inst.UpdatedAt = now
+
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer tx.Rollback(ctx)
+
+	// 先 bump revision（乐观锁），冲突即回滚，不写 plan
+	var newRev int
+	err = tx.QueryRow(ctx, `
+		UPDATE projects SET workflow_plan_revision = workflow_plan_revision + 1, updated_at = now()
+		WHERE id = $1 AND workflow_plan_revision = $2
+		RETURNING workflow_plan_revision
+	`, inst.ProjectID, expectedPlanRevision).Scan(&newRev)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, 0, ErrPlanRevisionConflict
+	}
+	if err != nil {
+		return nil, 0, err
+	}
+
+	var existingID int64
+	err = tx.QueryRow(ctx, `
+		SELECT id FROM project_workflow_instances WHERE project_id = $1 AND status = 'planning' LIMIT 1
+	`, inst.ProjectID).Scan(&existingID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		var nextNo int
+		if err := tx.QueryRow(ctx, `
+			SELECT COALESCE(MAX(instance_no), 0) + 1
+			FROM project_workflow_instances
+			WHERE project_id = $1
+		`, inst.ProjectID).Scan(&nextNo); err != nil {
+			return nil, 0, err
+		}
+		inst.InstanceNo = nextNo
+		err = tx.QueryRow(ctx, `
+			INSERT INTO project_workflow_instances (
+				project_id, workflow_version_id, instance_no, status, config_revision, plan_hash,
+				config_snapshot, created_by, created_at, updated_at
+			) VALUES ($1, $2, $3, 'planning', $4, $5, $6, $7, $8, $9)
+			RETURNING id
+		`, inst.ProjectID, inst.WorkflowVersionID, nextNo, inst.ConfigRevision, inst.PlanHash, snapRaw,
+			inst.CreatedBy, inst.CreatedAt, inst.UpdatedAt).Scan(&inst.ID)
+		if err != nil {
+			return nil, 0, err
+		}
+	} else if err != nil {
+		return nil, 0, err
+	} else {
+		err = tx.QueryRow(ctx, `
+			UPDATE project_workflow_instances SET
+				workflow_version_id = $2, config_revision = $3, plan_hash = $4, config_snapshot = $5,
+				updated_at = $6
+			WHERE id = $1 AND status = 'planning'
+			RETURNING id
+		`, existingID, inst.WorkflowVersionID, inst.ConfigRevision, inst.PlanHash, snapRaw, inst.UpdatedAt).Scan(&inst.ID)
+		if err != nil {
+			return nil, 0, err
+		}
+	}
+
+	if err := s.replaceInstanceChildren(ctx, tx, inst); err != nil {
+		return nil, 0, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, 0, err
+	}
+	return inst, newRev, nil
+}
+
+func (s *pgStore) replaceInstanceChildren(ctx context.Context, tx pgx.Tx, inst *WorkflowInstance) error {
+	_, err := tx.Exec(ctx, `
+		DELETE FROM document_duty_assignments WHERE doc_task_instance_id IN (
+			SELECT dti.id FROM project_doc_task_instances dti
+			JOIN project_stage_instances si ON si.id = dti.stage_instance_id
+			WHERE si.workflow_instance_id = $1
+		)
+	`, inst.ID)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `
+		DELETE FROM project_doc_task_instances WHERE stage_instance_id IN (
+			SELECT id FROM project_stage_instances WHERE workflow_instance_id = $1
+		)
+	`, inst.ID)
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec(ctx, `DELETE FROM project_stage_instances WHERE workflow_instance_id = $1`, inst.ID)
+	if err != nil {
+		return err
+	}
+
+	for i, st := range inst.ConfigSnapshot.Stages {
+		position := (i + 1) * 10
+		var stageID int64
+		err = tx.QueryRow(ctx, `
+			INSERT INTO project_stage_instances (workflow_instance_id, stage_key, position, enabled, status)
+			VALUES ($1, $2, $3, $4, 'pending') RETURNING id
+		`, inst.ID, st.StageKey, position, st.Enabled).Scan(&stageID)
+		if err != nil {
+			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, $4) RETURNING id
+			`, stageID, dt.TaskKey, dt.Enabled, status).Scan(&docTaskID)
+			if err != nil {
+				return err
+			}
+			if !dt.Enabled {
+				continue
+			}
+			mode, effective, _ := ResolveDocTaskDuty(dt)
+			if mode == DutyModeNone {
+				continue
+			}
+			for role, uid := range map[string]int64{
+				DutyAuthor: effective.AuthorUserID, DutyReviewer: effective.ReviewerUserID, DutyApprover: effective.ApproverUserID,
+			} {
+				if uid <= 0 {
+					continue
+				}
+				_, err = tx.Exec(ctx, `
+					INSERT INTO document_duty_assignments (doc_task_instance_id, duty_role, user_id, assigned_by)
+					VALUES ($1, $2, $3, $4)
+				`, docTaskID, role, uid, inst.CreatedBy)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}

internal/projectinstance/store_pg_token.go

@@ -0,0 +1,60 @@
+package projectinstance
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+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, policyRevision *int64) 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, author_only_policy_revision)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+	`, tokenHash, projectID, planHash, expectedVersion, raw, createdBy, expiresAt, policyRevision)
+	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
+	var expiresAt time.Time
+	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
+	`, tokenHash, projectID).Scan(&storedHash, &snapRaw, &usedAt, &expiresAt)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrTokenInvalid
+	}
+	if err != nil {
+		return nil, err
+	}
+	if usedAt != nil || time.Now().After(expiresAt) || storedHash != planHash {
+		return nil, ErrTokenInvalid
+	}
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE workflow_activation_tokens SET used_at = now() WHERE token = $1 AND used_at IS NULL
+	`, tokenHash)
+	if err != nil {
+		return nil, err
+	}
+	if tag.RowsAffected() == 0 {
+		return nil, ErrTokenInvalid
+	}
+	var plan WorkflowPlan
+	_ = json.Unmarshal(snapRaw, &plan)
+	return &plan, nil
+}

internal/projectinstance/store_seed.go

@@ -0,0 +1,20 @@
+package projectinstance
+
+import (
+	"fmt"
+)
+
+// SeedProjectMeta 供测试注入项目元数据。
+func SeedProjectMeta(store Store, projectID int64, lifecycle, activeKey string, selectionVer, planRev int) error {
+	ms, ok := store.(*memStore)
+	if !ok {
+		return fmt.Errorf("SeedProjectMeta only supported on memStore")
+	}
+	ms.mu.Lock()
+	defer ms.mu.Unlock()
+	ms.meta[projectID] = projectMeta{
+		lifecycle: lifecycle, activeKey: activeKey,
+		selectionVer: selectionVer, planRevision: planRev,
+	}
+	return nil
+}