From 4b168854ba41ad34e06cc3de424e8c8a3bfb6ec6 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 12 Sep 2026 05:51:58 +0000 Subject: [PATCH] feat(abac): filter blended squad context and add PG governance E2E Route BuildBlendedSquadContext through stage access gate, and add PG integration test covering activate, doc approval hook, notifications, ABAC, and stage advance. Co-authored-by: Cursor --- ...-12-piagent-e2e-bizdev-abac-test-report.md | 39 ++- internal/bizdev/service.go | 2 +- internal/bizdev/stage_access.go | 11 + internal/bizdev/stage_access_test.go | 62 ++++ internal/ctxmgr/manager.go | 15 +- .../server/workflow_governance_pg_test.go | 300 ++++++++++++++++++ 6 files changed, 423 insertions(+), 6 deletions(-) create mode 100644 internal/server/workflow_governance_pg_test.go 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) + } + +}