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 <cursoragent@cursor.com>
This commit is contained in:
weichen
committed by
weichen
weichen
co-authored by
Cursor
Cursor
parent
79811aa759
commit
4b168854ba
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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{})
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user