Agent-System/internal/projectinstance/pg_integration_test.go

package projectinstance_test

import (
	"context"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"chat-agent-system/internal/db"
	"chat-agent-system/internal/docgov"
	"chat-agent-system/internal/docrevision"
	"chat-agent-system/internal/notification"
	"chat-agent-system/internal/projectinstance"
	"chat-agent-system/internal/projectworkflow"
	"chat-agent-system/internal/workflowdef"
)

func openPGPool(t *testing.T) *pgxpool.Pool {
	t.Helper()
	dsn := db.SkipIfNoTestDatabaseURL(t)
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		t.Skip(err)
	}
	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		t.Skip(err)
	}
	migCtx, migCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer migCancel()
	if err := db.RunMigrations(migCtx, pool, db.ResolveMigrationsDir()); err != nil {
		pool.Close()
		t.Skipf("migration failed: %v", err)
	}
	if !projectinstance.TablesReady(ctx, pool) {
		pool.Close()
		t.Skip("project_workflow_instances not migrated")
	}
	if !workflowdef.TablesReady(ctx, pool) {
		pool.Close()
		t.Skip("workflow_definitions not migrated")
	}
	return pool
}

func TestPG_ActivationAndStageAdvance(t *testing.T) {
	pool := openPGPool(t)
	defer pool.Close()
	ctx := context.Background()

	// 治理链路仅接受官方工作流（BuildDefaultPlan official-only），seed 官方 main manifest。
	wfKey := workflowdef.OfficialKeyMain
	def, err := workflowdef.LoadManifestDefinition(workflowdef.V2ManifestFiles[wfKey])
	if err != nil {
		t.Fatalf("load manifest: %v", err)
	}
	wfStore := workflowdef.NewStore(pool)
	if err := wfStore.UpsertPublishedSeed(ctx, workflowdef.CreateDefinitionInput{
		Key:        wfKey,
		Name:       workflowdef.OfficialWorkflowName(wfKey),
		Definition: def,
	}, 1); err != nil {
		t.Fatalf("seed official workflow: %v", err)
	}

	var authorID, reviewerID, approverID int64
	rows, err := pool.Query(ctx, `SELECT id FROM users ORDER BY id ASC LIMIT 3`)
	if err != nil {
		t.Fatalf("list users: %v", err)
	}
	defer rows.Close()
	ids := make([]int64, 0, 3)
	for rows.Next() {
		var id int64
		if scanErr := rows.Scan(&id); scanErr != nil {
			t.Fatalf("scan user: %v", scanErr)
		}
		ids = append(ids, id)
	}
	if len(ids) < 3 {
		t.Skip("need at least 3 user rows for duty assignees")
	}
	authorID, reviewerID, approverID = ids[0], ids[1], ids[2]

	pwStore := projectworkflow.NewStore(pool)
	instStore := projectinstance.NewStore(pool)
	docChecker := projectinstance.NewDocGovChecker(docgov.NewGovernanceEngine(pool))
	svc := projectinstance.NewService(instStore, wfStore, pwStore, docChecker)

	var projectID int64
	code := "pg-p3-" + time.Now().Format("150405.000000")
	err = pool.QueryRow(ctx, `
		INSERT INTO projects (code, name, status, creator_user_id, lifecycle_status, active_workflow_key, workflow_selection_version, workflow_plan_revision)
		VALUES ($1, $2, 'active', $3, 'planning', '', 1, 0) RETURNING id
	`, code, "PG Phase3 Test "+code, authorID).Scan(&projectID)
	if err != nil {
		t.Fatalf("insert project: %v", err)
	}
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID)
	})

	plan, err := svc.BuildDefaultPlan(ctx, wfKey, 0)
	if err != nil {
		t.Fatalf("default plan: %v", err)
	}
	intakeTask := ""
	for i := range plan.Stages {
		for j := range plan.Stages[i].DocTasks {
			tk := plan.Stages[i].DocTasks[j].TaskKey
			plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{
				AuthorUserID: authorID, ReviewerUserID: reviewerID, ApproverUserID: approverID,
			}
			if tk == "intake" || tk == "bd_intake" {
				intakeTask = tk
			} else {
				// Keep a single blocking intake path so stage readiness is deterministic.
				plan.Stages[i].DocTasks[j].Enabled = false
			}
		}
	}
	if intakeTask == "" {
		t.Fatal("official main plan missing intake/bd_intake task")
	}

	_, err = svc.SaveWorkflowPlan(ctx, projectID, authorID, true, projectinstance.SavePlanInput{
		Plan: *plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1,
	})
	if err != nil {
		t.Fatalf("save plan: %v", err)
	}
	preview, err := svc.ActivationPreview(ctx, projectID, authorID, true)
	if err != nil {
		t.Fatalf("preview: %v", err)
	}
	activated, err := svc.Activate(ctx, projectID, authorID, true, projectinstance.ActivateInput{
		ConfirmationToken:      preview.ConfirmationToken,
		ExpectedProjectVersion: preview.ExpectedProjectVersion,
		ExpectedPlanRevision:   preview.ExpectedPlanRevision,
	})
	if err != nil {
		t.Fatalf("activate: %v", err)
	}

	docID := "doc_pg_p3_intake_" + code
	engine := docgov.NewGovernanceEngine(pool)
	_, err = engine.CreateOrGetDocument(ctx, docID, projectID, intakeTask, "客户需求采集", "V0.1", "", authorID, "张", reviewerID, "李", approverID, "王")
	if err != nil {
		t.Fatalf("create doc: %v", err)
	}
	_ = engine.MarkDocumentContentUpdated(ctx, docID, "hash_pg_p3", false)
	if _, err := engine.SubmitForReview(ctx, docID, authorID, "author", "张"); err != nil {
		t.Fatalf("submit for review: %v", err)
	}
	snapID, serr := docrevision.SeedContentSnapshotForTest(ctx, pool, docID, authorID, "# pg p3 intake")
	if serr != nil {
		t.Fatalf("seed snapshot: %v", serr)
	}
	if ferr := engine.FreezeHeadSnapshotForTest(ctx, docID, snapID); ferr != nil {
		t.Fatalf("freeze head snapshot: %v", ferr)
	}
	// 直写绕过了 docgov 引擎缓存；换新引擎使 approve 从 PG 装载快照。
	freshEngine := docgov.NewGovernanceEngine(pool)
	if _, err := freshEngine.ReviewApprove(ctx, docID, reviewerID, "reviewer", "李", "ok"); err != nil {
		t.Fatalf("review approve: %v", err)
	}
	_, err = freshEngine.ApproverApprove(ctx, docID, approverID, "approver", "王", "ok")
	if err != nil {
		t.Fatalf("approve doc: %v", err)
	}

	approvedResult, err := svc.HandleDocumentApproved(ctx, projectID, intakeTask)
	if err != nil {
		t.Fatalf("handle document approved: %v", err)
	}
	if approvedResult == nil || !approvedResult.StageReady {
		t.Fatalf("expected stage ready after approval, got %+v", approvedResult)
	}

	notifySvc := notification.NewService(pool)
	if notifySvc.Available() {
		created, emitErr := notifySvc.EmitStageReadyToAdvance(ctx, authorID, notification.StageReadyPayload{
			ProjectID:          projectID,
			WorkflowInstanceID: approvedResult.InstanceID,
			StageInstanceID:    approvedResult.StageInstanceID,
			StageKey:           approvedResult.StageKey,
			StageName:          approvedResult.StageName,
			ConfigRevision:     approvedResult.ConfigRevision,
			RequiredDocCount:   approvedResult.RequiredDocCount,
			ApprovedDocCount:   approvedResult.ApprovedDocCount,
		})
		if emitErr != nil {
			t.Fatalf("emit stage ready notification: %v", emitErr)
		}
		if !created {
			t.Fatal("expected stage ready notification to be created")
		}
	}

	stageKey := approvedResult.StageKey
	if stageKey == "" {
		stageKey = "V0.1_triage"
	}
	readiness, err := svc.GetStageReadiness(ctx, activated.Instance.ID, stageKey, authorID, true)
	if err != nil {
		t.Fatalf("readiness: %v", err)
	}
	if !readiness.Ready {
		otherTasks := []string{}
		for _, dt := range readiness.DocTasks {
			if !dt.Approved {
				otherTasks = append(otherTasks, dt.TaskKey+":"+dt.DocState)
			}
		}
		t.Fatalf("expected ready stage, pending=%v issues=%+v", otherTasks, readiness.BlockingIssues)
	}

	advanced, err := svc.AdvanceStage(ctx, activated.Instance.ID, authorID, true, projectinstance.AdvanceStageInput{
		StageKey: stageKey, ExpectedConfigRevision: activated.Instance.ConfigRevision,
	})
	if err != nil {
		t.Fatalf("advance: %v", err)
	}
	if advanced.NextStageKey == "" && !advanced.InstanceDone {
		t.Fatalf("expected next stage or done, got %+v", advanced)
	}
}