From 675b00e06ddbe3c5a8344ff9e876e0cef9b99d6d Mon Sep 17 00:00:00 2001 From: weichen Date: Tue, 22 Sep 2026 18:15:29 +0800 Subject: [PATCH] test: align official bd_intake task key and scoped task-run fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Official main.v2 uses bd_intake (not intake); stage gates stayed closed when fixtures approved the wrong task key. Task-run finalizers also require a real project_id for outbox event scope—seed scoped projects in PG fixtures. Co-authored-by: Cursor --- .../server/chat_history_refresh_pg_test.go | 6 ++--- .../server/phase5_fault_injection_pg_test.go | 2 +- .../server/phase5_finalize_fault_pg_test.go | 3 ++- .../server/task_run_persistence_pg_test.go | 6 +++-- internal/server/task_run_resume_pg_test.go | 7 +++--- internal/server/test_helpers_test.go | 24 +++++++++++++++++++ .../server/workflow_governance_pg_test.go | 2 +- 7 files changed, 39 insertions(+), 11 deletions(-) diff --git a/internal/server/chat_history_refresh_pg_test.go b/internal/server/chat_history_refresh_pg_test.go index e3df7c67..dd301a8f 100644 --- a/internal/server/chat_history_refresh_pg_test.go +++ b/internal/server/chat_history_refresh_pg_test.go @@ -113,14 +113,14 @@ func TestChatHistoryRefresh_PG_PreActivationAndTaskScope(t *testing.T) { } instSvc := projectinstance.NewService(projectinstance.NewStore(pool), nil, nil, nil) - taskScope, err := instSvc.ResolveMessageScopeForTask(ctx, projectID, "intake") + taskScope, err := instSvc.ResolveMessageScopeForTask(ctx, projectID, "bd_intake") if err != nil { t.Fatalf("resolve task scope: %v", err) } - taskUserMeta := map[string]any{"project_id": projectID, "task_key": "intake", "client_turn_id": "turn-doc-1"} + taskUserMeta := map[string]any{"project_id": projectID, "task_key": "bd_intake", "client_turn_id": "turn-doc-1"} projectinstance.ApplyScopeToMetadata(taskUserMeta, taskScope) _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, creatorID, "user", "请生成 intake 文档", "TRIGGER_TASK", "", taskUserMeta) - taskAsstMeta := map[string]any{"project_id": projectID, "task_key": "intake", "task_status": "completed"} + taskAsstMeta := map[string]any{"project_id": projectID, "task_key": "bd_intake", "task_status": "completed"} projectinstance.ApplyScopeToMetadata(taskAsstMeta, taskScope) _, _ = ctxMgr.RecordUserMessageWithID(ctx, sessID, 0, "assistant", "文档已生成", "TRIGGER_TASK", "", taskAsstMeta) diff --git a/internal/server/phase5_fault_injection_pg_test.go b/internal/server/phase5_fault_injection_pg_test.go index 87632cb6..50e2c876 100644 --- a/internal/server/phase5_fault_injection_pg_test.go +++ b/internal/server/phase5_fault_injection_pg_test.go @@ -102,7 +102,7 @@ func setupPGProjectReadyToAdvance(t *testing.T, pool *pgxpool.Pool, handler http docID := fmt.Sprintf("doc_p5_fault_%d", projectID) docEngine := bizdev.Snapshot(srv.BizdevService()).Gov - _, _ = docEngine.CreateOrGetDocument(ctx, docID, projectID, "intake", "Intake", "V0.1", "", creatorID, "A", reviewerID, "R", approverID, "P") + _, _ = docEngine.CreateOrGetDocument(ctx, docID, projectID, "bd_intake", "Intake", "V0.1", "", creatorID, "A", reviewerID, "R", approverID, "P") _ = docEngine.MarkDocumentContentUpdated(ctx, docID, "hash_p5_fault", false) _, _ = docEngine.SubmitForReview(ctx, docID, creatorID, "author", "Author") _, _ = docEngine.ReviewApprove(ctx, docID, reviewerID, "reviewer", "Reviewer", "ok") diff --git a/internal/server/phase5_finalize_fault_pg_test.go b/internal/server/phase5_finalize_fault_pg_test.go index e0d8bb8b..f4c6deaf 100644 --- a/internal/server/phase5_finalize_fault_pg_test.go +++ b/internal/server/phase5_finalize_fault_pg_test.go @@ -48,6 +48,7 @@ func TestPhase5_RollingRestartReplay_PG(t *testing.T) { handler, srv := setupPGGovernanceServer(t, pool) token := loginTestUser(t, handler, "zhanggong", "123456") userID := fetchAuthUserID(t, handler, token) + projectID := insertScopedTestProject(t, pool, userID, "pg-rolling") store := taskrun.NewPGStore(pool) runID := "run_rolling_" + time.Now().Format("150405.000000") @@ -66,7 +67,7 @@ func TestPhase5_RollingRestartReplay_PG(t *testing.T) { }) record := &taskrun.RunRecord{ - RunID: runID, SessionID: "sess_roll", UserID: userID, TaskKey: "intake", Status: "running", + RunID: runID, SessionID: "sess_roll", UserID: userID, ProjectID: projectID, TaskKey: "intake", Status: "running", } _, _ = srv.TaskRuns().CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") t.Cleanup(func() { diff --git a/internal/server/task_run_persistence_pg_test.go b/internal/server/task_run_persistence_pg_test.go index 2f715c67..89a9bbd4 100644 --- a/internal/server/task_run_persistence_pg_test.go +++ b/internal/server/task_run_persistence_pg_test.go @@ -25,6 +25,8 @@ func TestTaskRun_PersistenceAndReplayAfterRestart_PG(t *testing.T) { handler, srv := setupPGGovernanceServer(t, pool) token := loginTestUser(t, handler, "zhanggong", "123456") + userID := fetchAuthUserID(t, handler, token) + projectID := insertScopedTestProject(t, pool, userID, "pg-persist") runID := "run_pg_persist_" + time.Now().Format("150405") store := taskrun.NewPGStore(pool) @@ -45,8 +47,8 @@ func TestTaskRun_PersistenceAndReplayAfterRestart_PG(t *testing.T) { }) record := &taskrun.RunRecord{ - RunID: runID, SessionID: "sess_pg_persist", ProjectID: 0, - UserID: fetchAuthUserID(t, handler, token), TaskKey: "intake", Status: "running", + RunID: runID, SessionID: "sess_pg_persist", ProjectID: projectID, + UserID: userID, TaskKey: "intake", Status: "running", } _, err := srv.TaskRuns().CreateAndStart(context.Background(), runID, record, bizdev.TaskRunRequest{TaskKey: "intake"}, bizdev.StageRunRequest{}, "") if err != nil { diff --git a/internal/server/task_run_resume_pg_test.go b/internal/server/task_run_resume_pg_test.go index 3ce5f879..8a62b7a0 100644 --- a/internal/server/task_run_resume_pg_test.go +++ b/internal/server/task_run_resume_pg_test.go @@ -41,6 +41,7 @@ func TestTaskRun_ResumeInterrupted_PG(t *testing.T) { handler := srv.Routes() token := loginTestUser(t, handler, "zhanggong", "123456") userID := fetchAuthUserID(t, handler, token) + projectID := insertScopedTestProject(t, pool, userID, "pg-resume") workDir := filepath.Join(cfg.DataDir, "workspaces", "bizdev", "resume_test") if err := os.MkdirAll(workDir, 0o755); err != nil { @@ -52,9 +53,9 @@ func TestTaskRun_ResumeInterrupted_PG(t *testing.T) { runID := "run_resume_" + time.Now().Format("150405.000000") _, err := pool.Exec(context.Background(), ` INSERT INTO task_runs ( - run_id, session_id, user_id, task_key, status, workspace_id, error_message, created_at, updated_at - ) VALUES ($1, 'sess_resume', $2, 'intake', 'interrupted', $3, 'worker lease expired', now(), now()) - `, runID, userID, workDir) + run_id, session_id, project_id, user_id, task_key, status, workspace_id, error_message, created_at, updated_at + ) VALUES ($1, 'sess_resume', $2, $3, 'intake', 'interrupted', $4, 'worker lease expired', now(), now()) + `, runID, projectID, userID, workDir) if err != nil { t.Fatalf("seed interrupted run: %v", err) } diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 690f98fc..9259c862 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -162,3 +162,27 @@ func seedOfficialWorkflowManifest(t *testing.T, pool *pgxpool.Pool, workflowKey t.Fatalf("seed official workflow %s: %v", workflowKey, err) } } + +// insertScopedTestProject creates a project row so task-run finalizers can emit +// outbox events (Environment + ProjectID are both required by events.Validate). +func insertScopedTestProject(t *testing.T, pool *pgxpool.Pool, creatorID int64, namePrefix string) int64 { + t.Helper() + if creatorID <= 0 { + creatorID = 1 + } + code := namePrefix + "-" + time.Now().Format("150405.000000") + var projectID int64 + err := pool.QueryRow(context.Background(), ` + 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, namePrefix+" "+code, creatorID).Scan(&projectID) + if err != nil { + t.Fatalf("insert scoped project: %v", err) + } + _, _ = pool.Exec(context.Background(), ` + INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at) + VALUES ($1, $2, 'owner', 'active', now(), now()) ON CONFLICT DO NOTHING + `, projectID, creatorID) + t.Cleanup(func() { cleanupProject(t, pool, projectID) }) + return projectID +} diff --git a/internal/server/workflow_governance_pg_test.go b/internal/server/workflow_governance_pg_test.go index f6a4b340..ed4ef34f 100644 --- a/internal/server/workflow_governance_pg_test.go +++ b/internal/server/workflow_governance_pg_test.go @@ -220,7 +220,7 @@ func TestWorkflowGovernance_PG_ActivateApproveAdvanceChain(t *testing.T) { // 4. 文档批准链（经 bizdev DocGov 引擎，触发 onWorkflowDocumentApproved hook） docID := fmt.Sprintf("doc_pg_p5_intake_%d", projectID) docEngine := bizdev.Snapshot(srv.BizdevService()).Gov - _, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "intake", "Intake Doc", "V0.1", "", creatorID, "Author", reviewerID, "Reviewer", approverID, "Approver") + _, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "bd_intake", "Intake Doc", "V0.1", "", creatorID, "Author", reviewerID, "Reviewer", approverID, "Approver") if err != nil { t.Fatalf("create doc: %v", err) }