diff --git a/.doc/test-reports/2026-09-12-piagent-e2e-bizdev-abac-test-report.md b/.doc/test-reports/2026-09-12-piagent-e2e-bizdev-abac-test-report.md index 005cb1a2..e2e2dc2a 100644 --- a/.doc/test-reports/2026-09-12-piagent-e2e-bizdev-abac-test-report.md +++ b/.doc/test-reports/2026-09-12-piagent-e2e-bizdev-abac-test-report.md @@ -97,8 +97,41 @@ ok chat-agent-system/internal/bizdev 0.003s TestEnsureTaskStageAccess_De | bizdev Agent 上下文 ABAC | ✅ 已实现并注入 server | | 多角色 docgov Bearer 鉴权 | ✅ E2E 验证通过 | +### 用例 3：BlendedSquadContext 阶段 ABAC 过滤 + +- **Input**: `go test ./internal/bizdev/... -run TestBuildBlendedSquadContextForUser_FiltersByStage -v` +- **Actual Output**: + ``` + --- PASS: TestBuildBlendedSquadContextForUser_FiltersByStage (0.00s) + PASS + ``` +- **Verdict**: **PASS**（里程碑决策按 `stage_key` 过滤，仅保留当前允许阶段） + +### 用例 4：Phase 5 PG 治理全链路（activate → approve → advance） + +- **Input**: `go test ./internal/server/ -run TestWorkflowGovernance_PG -v` +- **Actual Output**: + ``` + --- PASS: TestWorkflowGovernance_PG_ActivateApproveAdvanceChain (0.36s) + PASS + ok chat-agent-system/internal/server 0.367s + ``` +- **Verdict**: **PASS**（含 stage_ready 通知、访客 403 ABAC、创建者阶段推进） +- **Duration**: ~0.36s + +## 七、 测试结论与后续优化建议 (Conclusion & Next Steps) + +**结论**：pi-agent E2E、bizdev ABAC、Blended 上下文过滤与 PG 治理链均已验收通过。 + +| 交付项 | 状态 | +|--------|------| +| pi-agent 1h E2E 全链路 | ✅ PASS（676s） | +| bizdev Agent 上下文 ABAC | ✅ | +| BuildBlendedSquadContext ABAC | ✅ | +| PG activate→approve→advance | ✅ | + **后续建议**： -1. 将 `BuildBlendedSquadContext` 进一步接入 ABAC，避免 squad 层混合上下文绕过 gate。 -2. 增加 PG live 实例下的 bizdev gate 集成测试（与 `TestResolveStageAccess_*` 联动）。 -3. Phase 5：workflow activate → doc approve → stage advance 的真实 Agent 治理全链路 E2E。 +1. 修复 PG 环境下 `POST /api/v1/projects` 中文名称 UTF-8 插入异常（当前 PG E2E 使用 SQL 直写规避）。 +2. Phase 5 扩展：结合 pi-agent 真实 Agent 的 workflow 治理全链路（在 PG + stub 链 PASS 基础上叠加）。 +3. 生产验收矩阵：权限负例、并发幂等、备份恢复演练。 diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index f4a72482..43e07d2c 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -589,7 +589,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun } if s.ctxMgr != nil { - blended, err := s.ctxMgr.BuildBlendedSquadContext(ctx, req.ProjectID, req.UserID, req.TaskKey, "squad") + blended, err := s.buildBlendedSquadContextForUser(ctx, req.ProjectID, req.UserID, req.TaskKey, "squad") if err == nil && blended != nil { if userCtxStr == "" && blended.PersonalizedGuidance != "" { userCtxStr = blended.PersonalizedGuidance diff --git a/internal/bizdev/stage_access.go b/internal/bizdev/stage_access.go index 0e915a85..5a65ebfc 100644 --- a/internal/bizdev/stage_access.go +++ b/internal/bizdev/stage_access.go @@ -48,3 +48,14 @@ func (s *Service) ensureTaskStageAccess(ctx context.Context, projectID, userID i } return s.stageGate.AllowTaskExecution(ctx, projectID, userID, taskKey) } + +func (s *Service) buildBlendedSquadContextForUser(ctx context.Context, projectID, userID int64, taskKey, role string) (*ctxmgr.BlendedSquadContext, error) { + if s == nil || s.ctxMgr == nil || projectID <= 0 { + return nil, nil + } + projCtx, err := s.getProjectContextForUser(ctx, projectID, userID) + if err != nil { + return nil, err + } + return s.ctxMgr.BuildBlendedSquadContextFromProject(ctx, projCtx, userID, taskKey, role) +} diff --git a/internal/bizdev/stage_access_test.go b/internal/bizdev/stage_access_test.go index 7cb99c19..5f53de2f 100644 --- a/internal/bizdev/stage_access_test.go +++ b/internal/bizdev/stage_access_test.go @@ -3,6 +3,7 @@ package bizdev import ( "context" "errors" + "strings" "testing" "chat-agent-system/internal/ctxmgr" @@ -46,6 +47,67 @@ func TestEnsureTaskStageAccess_Denied(t *testing.T) { } } +type stageFilteringGate struct { + mockStageGate + allowedStage string +} + +func (g *stageFilteringGate) AllowProjectContextRead(_ context.Context, _, _ int64) error { + return nil +} + +func (g *stageFilteringGate) FilterProjectContext(_ context.Context, _, _ int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext { + if pCtx == nil { + return pCtx + } + out := *pCtx + out.MilestoneDecisions = filterTestActivitiesByStage(pCtx.MilestoneDecisions, g.allowedStage) + out.AllActivities = filterTestActivitiesByStage(pCtx.AllActivities, g.allowedStage) + return &out +} + +func filterTestActivitiesByStage(items []ctxmgr.ActivityItem, stageKey string) []ctxmgr.ActivityItem { + out := make([]ctxmgr.ActivityItem, 0, len(items)) + for _, it := range items { + sk, _ := it.Detail["stage_key"].(string) + if sk == "" || sk == stageKey { + out = append(out, it) + } + } + return out +} + +func TestBuildBlendedSquadContextForUser_FiltersByStage(t *testing.T) { + ctx := context.Background() + mgr := ctxmgr.NewContextManager(nil, nil, ctxmgr.DefaultTimeDecayConfig()) + projectID := int64(42) + userID := int64(2) + + _, _ = mgr.RecordActivity(ctx, projectID, userID, "张工", "decision", "立项拍板", map[string]any{"stage_key": "biz_intake"}, 1, true) + _, _ = mgr.RecordActivity(ctx, projectID, userID, "张工", "decision", "研发阶段决策", map[string]any{"stage_key": "biz_feasibility"}, 1, true) + + svc := &Service{ + ctxMgr: mgr, + stageGate: &stageFilteringGate{ + allowedStage: "biz_intake", + }, + } + + blended, err := svc.buildBlendedSquadContextForUser(ctx, projectID, userID, "intake", "squad") + if err != nil { + t.Fatalf("build blended: %v", err) + } + if blended == nil { + t.Fatal("expected blended context") + } + if len(blended.KeyDecisions) != 1 { + t.Fatalf("expected 1 filtered decision, got %d: %v", len(blended.KeyDecisions), blended.KeyDecisions) + } + if !strings.Contains(blended.KeyDecisions[0], "立项拍板") { + t.Fatalf("unexpected decision: %v", blended.KeyDecisions) + } +} + func TestSetStageAccessGate_NilSafe(t *testing.T) { var svc *Service svc.SetStageAccessGate(&mockStageGate{}) diff --git a/internal/ctxmgr/manager.go b/internal/ctxmgr/manager.go index e5b258c1..26efa618 100644 --- a/internal/ctxmgr/manager.go +++ b/internal/ctxmgr/manager.go @@ -561,11 +561,22 @@ func (m *ContextManager) BuildBlendedSquadContext(ctx context.Context, projectID if err != nil { return nil, fmt.Errorf("get project context failed: %w", err) } + return m.BuildBlendedSquadContextFromProject(ctx, projCtx, triggerUserID, taskKey, role) +} + +// BuildBlendedSquadContextFromProject 基于已过滤的项目上下文装配混合 Agent 上下文（供 ABAC gate 注入后使用）。 +func (m *ContextManager) BuildBlendedSquadContextFromProject(ctx context.Context, projCtx *ProjectCollectiveContext, triggerUserID int64, taskKey, role string) (*BlendedSquadContext, error) { + if projCtx == nil { + return nil, nil + } userPref, _ := m.GetUserPreference(ctx, triggerUserID) - guidance := m.reflector.GeneratePromptGuidance(userPref) + guidance := "" + if m.reflector != nil { + guidance = m.reflector.GeneratePromptGuidance(userPref) + } - var decisions []string + decisions := make([]string, 0, len(projCtx.MilestoneDecisions)) for _, d := range projCtx.MilestoneDecisions { decisions = append(decisions, fmt.Sprintf("[%s] %s (%s)", d.ActivityType, d.Title, d.UserName)) } diff --git a/internal/server/workflow_governance_pg_test.go b/internal/server/workflow_governance_pg_test.go new file mode 100644 index 00000000..7f301e29 --- /dev/null +++ b/internal/server/workflow_governance_pg_test.go @@ -0,0 +1,300 @@ +package server_test + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/db" + "chat-agent-system/internal/notification" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/server" + "chat-agent-system/internal/workflowdef" +) + +func openPGPoolWithMigrations(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("TEST_DATABASE_URL") + if dsn == "" { + dsn = "postgres://app:dev_secret_db_pass@127.0.0.1:5433/agent_system_dev?sslmode=disable" + } + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Skipf("postgres unavailable: %v", err) + return nil + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + t.Skipf("postgres ping failed: %v", err) + return nil + } + migCtx, migCancel := context.WithTimeout(context.Background(), 60*time.Second) + defer migCancel() + if err := db.RunMigrations(migCtx, pool, db.ResolveMigrationsDir()); err != nil { + pool.Close() + t.Skipf("migration failed: %v", err) + return nil + } + if !projectinstance.TablesReady(ctx, pool) || !workflowdef.TablesReady(ctx, pool) { + pool.Close() + t.Skip("workflow/project instance tables not ready") + } + return pool +} + +func fetchAuthUserID(t *testing.T, handler http.Handler, token string) int64 { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/v1/auth/me", nil) + withBearerAuth(req, token) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("auth/me: %d %s", rec.Code, rec.Body.String()) + } + var me struct { + User struct { + ID int64 `json:"id"` + } `json:"user"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &me); err != nil { + t.Fatalf("decode auth/me: %v", err) + } + if me.User.ID <= 0 { + t.Fatal("expected positive user id from auth/me") + } + return me.User.ID +} + +func setupPGGovernanceServer(t *testing.T, pool *pgxpool.Pool) (http.Handler, *server.Server) { + t.Helper() + cfg := &config.Config{ + Env: "dev", + DataDir: t.TempDir(), + AgentBackend: "stub", + AllowInMemory: false, + ContextDbHost: "127.0.0.1", + ContextDbPort: 5433, + ContextDbUser: "app", + ContextDbPassword: "dev_secret_db_pass", + ContextDbName: "agent_system_dev", + AdminUsername: "admin", + AdminPassword: "admin-secret-password-123", + } + 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 +} + +func TestWorkflowGovernance_PG_ActivateApproveAdvanceChain(t *testing.T) { + pool := openPGPoolWithMigrations(t) + if pool == nil { + return + } + defer pool.Close() + + ctx := context.Background() + handler, srv := setupPGGovernanceServer(t, pool) + + wfSvc := workflowdef.NewService(pool, nil) + wfKey := "pg_p5_gov_" + time.Now().Format("150405.000") + _, ver, err := wfSvc.CreateDefinition(ctx, workflowdef.CreateDefinitionInput{ + Key: wfKey, Name: "PG Phase5 Governance WF", + Definition: workflowdef.VersionDefinition{ + SchemaVersion: 1, + Stages: []workflowdef.StageDef{ + { + Key: "biz_intake", Name: "商务摸底", Position: 0, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "intake", Required: true, Blocking: true, DefaultEnabled: true, + OutputDocType: "intake", OutputDocTitle: "客户需求采集", + }}, + }, + { + Key: "biz_feasibility", Name: "法规可行性", Position: 1, Optional: true, + DocTasks: []workflowdef.DocTaskDef{{ + Key: "feasibility_reg", Required: false, Blocking: false, DefaultEnabled: true, + OutputDocType: "feasibility", OutputDocTitle: "法规路径评估", + }}, + }, + }, + }, + }, 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") + guestToken := loginTestUser(t, handler, "guest", "123456") + creatorID := fetchAuthUserID(t, handler, creatorToken) + + // 1. 创建项目（PG 直写，与 projectinstance PG 集成测试一致） + var projectID int64 + code := "pg-p5-" + time.Now().Format("150405.000") + 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, 'PG Phase5 Governance Test', 'active', $2, 'planning', '', 1, 0) RETURNING id + `, code, creatorID).Scan(&projectID) + if err != nil { + t.Fatalf("insert project: %v", err) + } + _, err = pool.Exec(ctx, ` + INSERT INTO project_members (project_id, user_id, project_role, status, joined_at, updated_at) + VALUES ($1, $2, 'owner', 'active', now(), now()) + ON CONFLICT (project_id, user_id) DO NOTHING + `, projectID, creatorID) + if err != nil { + t.Fatalf("insert project member: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM projects WHERE id = $1`, projectID) + }) + // 2. 获取默认规划并保存三权 + 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 + if err := json.Unmarshal(recDefault.Body.Bytes(), &plan); err != nil { + t.Fatalf("decode plan: %v", err) + } + reviewerID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "lizhuguan", "123456")) + approverID := fetchAuthUserID(t, handler, loginTestUser(t, handler, "wangzong", "123456")) + 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()) + } + + // 3. 激活预览 + 确认激活 + 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) + if recPreview.Code != http.StatusOK { + t.Fatalf("activation preview: %d %s", recPreview.Code, recPreview.Body.String()) + } + var preview projectinstance.ActivationPreview + if err := json.Unmarshal(recPreview.Body.Bytes(), &preview); err != nil { + t.Fatalf("decode preview: %v", err) + } + 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 + if err := json.Unmarshal(recActivate.Body.Bytes(), &activated); err != nil { + t.Fatalf("decode activate: %v", err) + } + instanceID := activated.Instance.ID + + // 4. 文档批准链（经 bizdev DocGov 引擎，触发 onWorkflowDocumentApproved hook） + docID := fmt.Sprintf("doc_pg_p5_intake_%d", projectID) + docEngine := srv.BizdevService().DocGov() + _, err = docEngine.CreateOrGetDocument(ctx, docID, projectID, "intake", "Intake Doc", "V0.1", "", creatorID, "Author", reviewerID, "Reviewer", approverID, "Approver") + if err != nil { + t.Fatalf("create doc: %v", err) + } + _ = docEngine.MarkDocumentContentUpdated(ctx, docID, "hash_pg_p5", 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) + } + + // 5. 创建者应收到 stage ready 通知 + reqNotify := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/projects/%d/notifications", projectID), nil) + withBearerAuth(reqNotify, creatorToken) + recNotify := httptest.NewRecorder() + handler.ServeHTTP(recNotify, reqNotify) + if recNotify.Code != http.StatusOK { + t.Fatalf("list notifications: %d %s", recNotify.Code, recNotify.Body.String()) + } + var notifyResp struct { + Items []map[string]any `json:"items"` + } + if err := json.Unmarshal(recNotify.Body.Bytes(), ¬ifyResp); err != nil { + t.Fatalf("decode notifications: %v body=%s", err, recNotify.Body.String()) + } + foundReady := false + for _, n := range notifyResp.Items { + if n["type"] == notification.TypeStageReadyToAdvance { + foundReady = true + break + } + } + if !foundReady { + t.Fatalf("expected stage_ready_to_advance notification, got %+v", notifyResp.Items) + } + + // 6. ABAC：访客无权读取项目上下文 + reqCtxGuest := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/api/v1/context/project?project_id=%d", projectID), nil) + withBearerAuth(reqCtxGuest, guestToken) + recCtxGuest := httptest.NewRecorder() + handler.ServeHTTP(recCtxGuest, reqCtxGuest) + if recCtxGuest.Code != http.StatusForbidden { + t.Fatalf("guest project context expected 403, got %d %s", recCtxGuest.Code, recCtxGuest.Body.String()) + } + + // 7. 创建者推进阶段 + 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 stage: %d %s", recAdvance.Code, recAdvance.Body.String()) + } + var advanceResult projectinstance.AdvanceStageResult + if err := json.Unmarshal(recAdvance.Body.Bytes(), &advanceResult); err != nil { + t.Fatalf("decode advance: %v", err) + } + if advanceResult.NextStageKey != "biz_feasibility" && !advanceResult.InstanceDone { + t.Fatalf("unexpected advance result: %+v", advanceResult) + } + +}