273 lines
9.9 KiB
Go
273 lines
9.9 KiB
Go
package server_test
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
|
||
"chat-agent-system/internal/bizdev"
|
||
"chat-agent-system/internal/notification"
|
||
"chat-agent-system/internal/projectinstance"
|
||
"chat-agent-system/internal/projectworkflow"
|
||
"chat-agent-system/internal/server"
|
||
"chat-agent-system/internal/squad"
|
||
"chat-agent-system/internal/workflowdef"
|
||
)
|
||
|
||
func piAgentRunnerHealthy() bool {
|
||
client := &http.Client{Timeout: 3 * time.Second}
|
||
resp, err := client.Get("http://127.0.0.1:18865/health")
|
||
if err != nil {
|
||
return false
|
||
}
|
||
resp.Body.Close()
|
||
return resp.StatusCode == http.StatusOK
|
||
}
|
||
|
||
func setupPGPiAgentServer(t *testing.T, pool *pgxpool.Pool) (http.Handler, *server.Server) {
|
||
t.Helper()
|
||
cfg := configFromTestDatabaseURL(t)
|
||
cfg.AgentBackend = "pi-agent"
|
||
cfg.PiAgentRunnerURL = "http://127.0.0.1:18865"
|
||
cfg.PiAgentModel = "MiniMax-M3"
|
||
cfg.PiAgentProvider = "minimax"
|
||
svc := bizdev.MustNewService(t, cfg)
|
||
if svc.Pool() == nil {
|
||
t.Skip("bizdev service not connected to postgres")
|
||
}
|
||
srv := server.NewServer(cfg, svc)
|
||
return srv.Routes(), srv
|
||
}
|
||
|
||
// TestWorkflowGovernance_PiAgent_PG_FullChain
|
||
// PG 治理链 + pi-agent 真实 Agent：activate → pi-agent intake → approve → advance
|
||
//
|
||
// go test ./internal/server/ -run TestWorkflowGovernance_PiAgent_PG -timeout 1h -v
|
||
func TestWorkflowGovernance_PiAgent_PG_FullChain(t *testing.T) {
|
||
if testing.Short() {
|
||
t.Skip("skipping slow pi-agent PG E2E in short mode")
|
||
}
|
||
if !piAgentRunnerHealthy() {
|
||
t.Skip("pi-agent runner not available at http://127.0.0.1:18865/health")
|
||
}
|
||
|
||
pool := openPGPoolWithMigrations(t)
|
||
if pool == nil {
|
||
return
|
||
}
|
||
defer pool.Close()
|
||
|
||
ctx := context.Background()
|
||
handler, srv := setupPGPiAgentServer(t, pool)
|
||
ts := httptest.NewServer(handler)
|
||
defer ts.Close()
|
||
client := &http.Client{Timeout: 50 * time.Minute}
|
||
|
||
wfSvc := workflowdef.NewService(pool, nil)
|
||
wfKey := "pg_pi_p5_" + time.Now().Format("150405.000")
|
||
_, ver, err := wfSvc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{
|
||
Key: wfKey, Name: "PG PiAgent Governance WF",
|
||
Definition: workflowdef.VersionDefinition{
|
||
SchemaVersion: 1,
|
||
Stages: []workflowdef.StageDef{{
|
||
Key: "biz_intake", Name: "Intake", Position: 0,
|
||
DocTasks: []workflowdef.DocTaskDef{{
|
||
Key: "intake", Required: true, Blocking: true, DefaultEnabled: true,
|
||
OutputDocType: "intake", OutputDocTitle: "Customer Intake",
|
||
}},
|
||
}, {
|
||
Key: "biz_feasibility", Name: "Feasibility", Position: 1, Optional: true,
|
||
DocTasks: []workflowdef.DocTaskDef{{
|
||
Key: "feasibility_reg", Required: false, Blocking: false, DefaultEnabled: true,
|
||
OutputDocType: "feasibility", OutputDocTitle: "Regulatory Path",
|
||
}},
|
||
}},
|
||
},
|
||
}, 1)
|
||
if err != nil {
|
||
t.Fatalf("create workflow def: %v", err)
|
||
}
|
||
if _, err := wfSvc.PublishVersion(ctx, ver.WorkflowID, ver.Version, 1); err != nil {
|
||
t.Fatalf("publish workflow: %v", err)
|
||
}
|
||
|
||
creatorToken := loginTestUser(t, handler, "zhanggong", "123456")
|
||
reviewerToken := loginTestUser(t, handler, "lizhuguan", "123456")
|
||
approverToken := loginTestUser(t, handler, "wangzong", "123456")
|
||
creatorID := fetchAuthUserID(t, handler, creatorToken)
|
||
reviewerID := fetchAuthUserID(t, handler, reviewerToken)
|
||
approverID := fetchAuthUserID(t, handler, approverToken)
|
||
|
||
// HTTP 创建中文名项目（验证 UTF-8 修复）
|
||
createBody, _ := json.Marshal(projectworkflow.CreateProjectInput{
|
||
Name: "PG PiAgent 治理链中文项目",
|
||
DeviceType: "Ultrasonic Device",
|
||
IntendedUse: "Tissue resection",
|
||
TargetMarket: "NMPA Class III",
|
||
})
|
||
reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(createBody))
|
||
withBearerAuth(reqCreate, creatorToken)
|
||
recCreate := httptest.NewRecorder()
|
||
handler.ServeHTTP(recCreate, reqCreate)
|
||
if recCreate.Code != http.StatusCreated {
|
||
t.Fatalf("create project: %d %s", recCreate.Code, recCreate.Body.String())
|
||
}
|
||
var proj projectworkflow.Project
|
||
if err := json.Unmarshal(recCreate.Body.Bytes(), &proj); err != nil {
|
||
t.Fatalf("decode project: %v", err)
|
||
}
|
||
t.Cleanup(func() { cleanupProject(t, pool, proj.ID) })
|
||
projectID := proj.ID
|
||
t.Logf("✓ [1. 项目创建] id=%d code=%s", projectID, proj.Code)
|
||
|
||
// 规划 + 激活
|
||
reqDefault := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/workflow-plan/default?workflow_key=%s&version=1", projectID, wfKey), nil)
|
||
withBearerAuth(reqDefault, creatorToken)
|
||
recDefault := httptest.NewRecorder()
|
||
handler.ServeHTTP(recDefault, reqDefault)
|
||
if recDefault.Code != http.StatusOK {
|
||
t.Fatalf("default plan: %d %s", recDefault.Code, recDefault.Body.String())
|
||
}
|
||
var plan projectinstance.WorkflowPlan
|
||
_ = json.Unmarshal(recDefault.Body.Bytes(), &plan)
|
||
for i := range plan.Stages {
|
||
for j := range plan.Stages[i].DocTasks {
|
||
plan.Stages[i].DocTasks[j].Duties = projectinstance.DocDuty{
|
||
AuthorUserID: creatorID, ReviewerUserID: reviewerID, ApproverUserID: approverID,
|
||
}
|
||
}
|
||
}
|
||
saveBody, _ := json.Marshal(projectinstance.SavePlanInput{Plan: plan, ExpectedPlanRevision: 0, ExpectedSelectionVersion: 1})
|
||
reqSave := httptest.NewRequest(http.MethodPut, fmt.Sprintf("/api/v1/projects/%d/workflow-plan", projectID), bytes.NewReader(saveBody))
|
||
withBearerAuth(reqSave, creatorToken)
|
||
recSave := httptest.NewRecorder()
|
||
handler.ServeHTTP(recSave, reqSave)
|
||
if recSave.Code != http.StatusOK {
|
||
t.Fatalf("save plan: %d %s", recSave.Code, recSave.Body.String())
|
||
}
|
||
|
||
reqPreview := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/activation-preview", projectID), nil)
|
||
withBearerAuth(reqPreview, creatorToken)
|
||
recPreview := httptest.NewRecorder()
|
||
handler.ServeHTTP(recPreview, reqPreview)
|
||
var preview projectinstance.ActivationPreview
|
||
_ = json.Unmarshal(recPreview.Body.Bytes(), &preview)
|
||
|
||
activateBody, _ := json.Marshal(projectinstance.ActivateInput{
|
||
ConfirmationToken: preview.ConfirmationToken, ExpectedProjectVersion: preview.ExpectedProjectVersion,
|
||
ExpectedPlanRevision: preview.ExpectedPlanRevision,
|
||
})
|
||
reqActivate := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/projects/%d/workflow-instances/activate", projectID), bytes.NewReader(activateBody))
|
||
withBearerAuth(reqActivate, creatorToken)
|
||
recActivate := httptest.NewRecorder()
|
||
handler.ServeHTTP(recActivate, reqActivate)
|
||
if recActivate.Code != http.StatusOK {
|
||
t.Fatalf("activate: %d %s", recActivate.Code, recActivate.Body.String())
|
||
}
|
||
var activated projectinstance.ActivateResult
|
||
_ = json.Unmarshal(recActivate.Body.Bytes(), &activated)
|
||
instanceID := activated.Instance.ID
|
||
t.Logf("✓ [2. 工作流激活] instance_id=%d", instanceID)
|
||
|
||
// pi-agent 执行 intake（精简单步 WBS）
|
||
taskStart := time.Now()
|
||
taskBody, _ := json.Marshal(server.TaskRunRequestDTO{
|
||
TaskKey: "intake",
|
||
ProjectName: proj.Name,
|
||
DeviceType: proj.DeviceType,
|
||
IntendedUse: "Tissue resection",
|
||
TargetMarket: "NMPA Class III",
|
||
Backend: "pi-agent",
|
||
ProjectID: projectID,
|
||
AuthorID: creatorID,
|
||
ReviewerID: reviewerID,
|
||
ApproverID: approverID,
|
||
CustomSteps: []squad.StepSpec{{
|
||
ID: "step_01_intake", Role: "writer", Title: "Intake snapshot",
|
||
SubWorkspace: "tasks/01_intake", Mode: "execute",
|
||
Accept: []string{"Deliver output.md with project intake summary"},
|
||
}},
|
||
})
|
||
reqTask, _ := http.NewRequest(http.MethodPost, ts.URL+"/api/v1/tasks/run", bytes.NewReader(taskBody))
|
||
withBearerAuth(reqTask, creatorToken)
|
||
respTask, err := client.Do(reqTask)
|
||
if err != nil {
|
||
t.Fatalf("pi-agent task run: %v", err)
|
||
}
|
||
defer respTask.Body.Close()
|
||
if respTask.StatusCode != http.StatusOK {
|
||
var buf bytes.Buffer
|
||
_, _ = buf.ReadFrom(respTask.Body)
|
||
t.Fatalf("pi-agent task HTTP %d: %s", respTask.StatusCode, buf.String())
|
||
}
|
||
var taskRes struct {
|
||
Success bool `json:"success"`
|
||
DocRecord *struct {
|
||
DocID string `json:"doc_id"`
|
||
} `json:"doc_record"`
|
||
ArtifactTitle string `json:"artifact_title"`
|
||
}
|
||
_ = json.NewDecoder(respTask.Body).Decode(&taskRes)
|
||
if !taskRes.Success {
|
||
t.Fatal("pi-agent intake task failed")
|
||
}
|
||
docID := ""
|
||
if taskRes.DocRecord != nil {
|
||
docID = taskRes.DocRecord.DocID
|
||
}
|
||
if docID == "" {
|
||
docID = fmt.Sprintf("doc_proj%d_intake", projectID)
|
||
}
|
||
t.Logf("✓ [3. pi-agent intake] doc=%s title=%s elapsed=%s", docID, taskRes.ArtifactTitle, time.Since(taskStart).Round(time.Second))
|
||
|
||
// 文档批准链
|
||
docEngine := srv.BizdevService().DocGov()
|
||
_ = docEngine.MarkDocumentContentUpdated(ctx, docID, "hash_pi_pg", false)
|
||
_, _ = docEngine.SubmitForReview(ctx, docID, creatorID, "author", "Author")
|
||
_, _ = docEngine.ReviewApprove(ctx, docID, reviewerID, "reviewer", "Reviewer", "ok")
|
||
_, err = docEngine.ApproverApprove(ctx, docID, approverID, "approver", "Approver", "ok")
|
||
if err != nil {
|
||
t.Fatalf("approve doc: %v", err)
|
||
}
|
||
t.Log("✓ [4. 文档批准] stage ready hook 已触发")
|
||
|
||
// 通知 + 阶段推进
|
||
reqNotify := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/notifications", projectID), nil)
|
||
withBearerAuth(reqNotify, creatorToken)
|
||
recNotify := httptest.NewRecorder()
|
||
handler.ServeHTTP(recNotify, reqNotify)
|
||
var notifyResp struct {
|
||
Items []map[string]any `json:"items"`
|
||
}
|
||
_ = json.Unmarshal(recNotify.Body.Bytes(), ¬ifyResp)
|
||
foundReady := false
|
||
for _, n := range notifyResp.Items {
|
||
if n["type"] == notification.TypeStageReadyToAdvance {
|
||
foundReady = true
|
||
break
|
||
}
|
||
}
|
||
if !foundReady {
|
||
t.Fatalf("expected stage_ready notification, got %+v", notifyResp.Items)
|
||
}
|
||
|
||
advanceBody, _ := json.Marshal(projectinstance.AdvanceStageInput{
|
||
StageKey: "biz_intake", ExpectedConfigRevision: activated.Instance.ConfigRevision,
|
||
})
|
||
reqAdvance := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/workflow-instances/%d/stages/biz_intake/advance", instanceID), bytes.NewReader(advanceBody))
|
||
withBearerAuth(reqAdvance, creatorToken)
|
||
recAdvance := httptest.NewRecorder()
|
||
handler.ServeHTTP(recAdvance, reqAdvance)
|
||
if recAdvance.Code != http.StatusOK {
|
||
t.Fatalf("advance: %d %s", recAdvance.Code, recAdvance.Body.String())
|
||
}
|
||
t.Log("✓ [5. 阶段推进] PG + pi-agent 治理全链路完成")
|
||
}
|