weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(abac): gate bizdev agent context by workflow stage duties

Browse Source 
Inject stage access gate into bizdev service so task execution, project context reads, and prior docs respect live workflow ABAC.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-13 02:50:23 +00:00
committed by weichen
co-authored by Cursor
parent f6410508aa
commit 7c5f8deeb1
5 changed files with 266 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/bizdev/service.go
+33 -6
View File
@@ -39,6 +39,7 @@ type Service struct {
projectWorkflow *projectworkflow.Service
pool *pgxpool.Pool
schemaVersion int64
stageGate StageAccessGate
}
// NewService 构造业务服务；生产/test 环境数据库不可用且未显式 ALLOW_IN_MEMORY 时返回 error。
@@ -230,6 +231,14 @@ func (s *Service) ListStages() []StageMeta {
}
func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRunResult, error) {
if err := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, req.TaskKey); err != nil {
return &TaskRunResult{
TaskKey: req.TaskKey,
Success: false,
ErrorReason: err.Error(),
}, err
}
materials, inputManifest, err := s.ResolveMaterials(ctx, req)
if err != nil {
return nil, err
@@ -240,7 +249,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
// 针对 system_reqs（智能产品经理）：若未显式提供 materials，自动抽取项目事实黑板/集体上下文为需求输入
if req.TaskKey == "system_reqs" && len(materials) == 0 && s.ctxMgr != nil && req.ProjectID > 0 {
coll, err := s.ctxMgr.GetProjectContext(ctx, req.ProjectID)
coll, err := s.getProjectContextForUser(ctx, req.ProjectID, req.UserID)
if err == nil && coll != nil {
var b strings.Builder
customer := coll.BlackboardFacts["customer_name"]
@@ -342,9 +351,11 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
if req.TaskKey == "test_manager" {
upstreamDocID := fmt.Sprintf("doc_proj%d_system_reqs", req.ProjectID)
if s.docGov != nil && req.ProjectID > 0 {
doc, err := s.docGov.GetDocument(ctx, upstreamDocID)
if err == nil && doc != nil && doc.CurrentState == docgov.StateApproved {
upstreamSRDoc = doc
if gateErr := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, "system_reqs"); gateErr == nil {
doc, err := s.docGov.GetDocument(ctx, upstreamDocID)
if err == nil && doc != nil && doc.CurrentState == docgov.StateApproved {
upstreamSRDoc = doc
}
}
}
if upstreamSRDoc == nil {
@@ -592,6 +603,11 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
}
}
priorDocs := req.PriorDocs
if s.stageGate != nil && req.ProjectID > 0 && req.UserID > 0 {
priorDocs = s.stageGate.FilterPriorDocs(ctx, req.ProjectID, req.UserID, priorDocs)
}
squadReq := squad.SquadRunRequest{
TaskKey: req.TaskKey,
Title: req.Title,
@@ -608,7 +624,7 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
CustomSteps: req.CustomSteps,
AutoRepair: req.AutoRepair,
KnowledgeHits: hits,
PriorDocs: req.PriorDocs,
PriorDocs: priorDocs,
UserContext: userCtxStr,
MemoryBankFiles: memBank,
SessionID: req.SessionID,
@@ -697,6 +713,14 @@ func (s *Service) ExecuteStage(ctx context.Context, req StageRunRequest) (*Stage
func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest) (*StageRunResult, error) {
start := time.Now()
if err := s.ensureTaskStageAccess(ctx, req.ProjectID, req.UserID, string(req.Stage)); err != nil {
return &StageRunResult{
Stage: req.Stage,
Success: false,
ErrorReason: err.Error(),
}, err
}
// 1. 确定工作区目录基准
stageKey := string(req.Stage)
if stageKey == "" {
@@ -732,7 +756,7 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest)
// 若存在项目协同管理器，优先载入已沉淀的项目事实与黑板
if s.ctxMgr != nil && req.ProjectID > 0 {
if projCtx, err := s.ctxMgr.GetProjectContext(ctx, req.ProjectID); err == nil && projCtx != nil {
if projCtx, err := s.getProjectContextForUser(ctx, req.ProjectID, req.UserID); err == nil && projCtx != nil {
for k, v := range projCtx.BlackboardFacts {
currentFields[k] = v
}
@@ -761,6 +785,9 @@ func (s *Service) executeStageInternal(ctx context.Context, req StageRunRequest)
}
accumulatedPriorDocs := append([]squad.PriorDocItem{}, req.PriorDocs...)
if s.stageGate != nil && req.ProjectID > 0 && req.UserID > 0 {
accumulatedPriorDocs = s.stageGate.FilterPriorDocs(ctx, req.ProjectID, req.UserID, accumulatedPriorDocs)
}
var executedArtifacts []*TaskRunResult
// 4. 按顺序执行当前阶段的子任务序列，并将每一轮产出作为后续任务的 PriorDoc 和事实补充
internal/bizdev/stage_access.go
+50
View File
@@ -0,0 +1,50 @@
package bizdev
import (
"context"
"errors"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/squad"
)
var ErrStageAccessDenied = errors.New("bizdev: workflow stage access denied")
// StageAccessGate 工作流阶段 ABAC 门禁（由 server 注入 projectinstance 实现）。
type StageAccessGate interface {
AllowTaskExecution(ctx context.Context, projectID, userID int64, taskKey string) error
AllowProjectContextRead(ctx context.Context, projectID, userID int64) error
FilterProjectContext(ctx context.Context, projectID, userID int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext
FilterPriorDocs(ctx context.Context, projectID, userID int64, docs []squad.PriorDocItem) []squad.PriorDocItem
}
func (s *Service) SetStageAccessGate(gate StageAccessGate) {
if s == nil {
return
}
s.stageGate = gate
}
func (s *Service) getProjectContextForUser(ctx context.Context, projectID, userID int64) (*ctxmgr.ProjectCollectiveContext, error) {
if s == nil || s.ctxMgr == nil || projectID <= 0 {
return nil, nil
}
coll, err := s.ctxMgr.GetProjectContext(ctx, projectID)
if err != nil {
return nil, err
}
if s.stageGate == nil || userID <= 0 {
return coll, nil
}
if err := s.stageGate.AllowProjectContextRead(ctx, projectID, userID); err != nil {
return nil, err
}
return s.stageGate.FilterProjectContext(ctx, projectID, userID, coll), nil
}
func (s *Service) ensureTaskStageAccess(ctx context.Context, projectID, userID int64, taskKey string) error {
if s == nil || s.stageGate == nil || projectID <= 0 || userID <= 0 {
return nil
}
return s.stageGate.AllowTaskExecution(ctx, projectID, userID, taskKey)
}
internal/bizdev/stage_access_test.go
+57
View File
@@ -0,0 +1,57 @@
package bizdev
import (
"context"
"errors"
"testing"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/squad"
)
type mockStageGate struct {
taskErr error
ctxErr error
filterCtx *ctxmgr.ProjectCollectiveContext
filtered []squad.PriorDocItem
}
func (m *mockStageGate) AllowTaskExecution(_ context.Context, _, _ int64, _ string) error {
return m.taskErr
}
func (m *mockStageGate) AllowProjectContextRead(_ context.Context, _, _ int64) error {
return m.ctxErr
}
func (m *mockStageGate) FilterProjectContext(_ context.Context, _, _ int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext {
if m.filterCtx != nil {
return m.filterCtx
}
return pCtx
}
func (m *mockStageGate) FilterPriorDocs(_ context.Context, _, _ int64, docs []squad.PriorDocItem) []squad.PriorDocItem {
if m.filtered != nil {
return m.filtered
}
return docs
}
func TestEnsureTaskStageAccess_Denied(t *testing.T) {
svc := &Service{stageGate: &mockStageGate{taskErr: ErrStageAccessDenied}}
err := svc.ensureTaskStageAccess(context.Background(), 1, 2, "intake")
if !errors.Is(err, ErrStageAccessDenied) {
t.Fatalf("expected ErrStageAccessDenied, got %v", err)
}
}
func TestSetStageAccessGate_NilSafe(t *testing.T) {
var svc *Service
svc.SetStageAccessGate(&mockStageGate{})
s := &Service{}
s.SetStageAccessGate(&mockStageGate{filtered: []squad.PriorDocItem{{DocKey: "intake"}}})
if s.stageGate == nil {
t.Fatal("expected gate to be set")
}
}
internal/server/bizdev_stage_gate.go
+123
View File
@@ -0,0 +1,123 @@
package server
import (
"context"
"fmt"
"chat-agent-system/internal/bizdev"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/projectinstance"
"chat-agent-system/internal/squad"
)
type bizdevStageGate struct {
instSvc *projectinstance.Service
pwSvc interface {
IsProjectCreator(ctx context.Context, projectID int64, userID int64) (bool, error)
}
}
func newBizdevStageGate(instSvc *projectinstance.Service, pwSvc interface {
IsProjectCreator(ctx context.Context, projectID int64, userID int64) (bool, error)
}) bizdev.StageAccessGate {
if instSvc == nil || !instSvc.Available() {
return nil
}
return &bizdevStageGate{instSvc: instSvc, pwSvc: pwSvc}
}
func (g *bizdevStageGate) access(ctx context.Context, projectID, userID int64) (*projectinstance.StageAccess, error) {
if g == nil || g.instSvc == nil || !g.instSvc.Available() || projectID <= 0 || userID <= 0 {
return &projectinstance.StageAccess{}, nil
}
isCreator := false
if g.pwSvc != nil {
isCreator, _ = g.pwSvc.IsProjectCreator(ctx, projectID, userID)
}
return g.instSvc.ResolveStageAccess(ctx, projectID, userID, isCreator)
}
func (g *bizdevStageGate) AllowTaskExecution(ctx context.Context, projectID, userID int64, taskKey string) error {
access, err := g.access(ctx, projectID, userID)
if err != nil {
return err
}
if access == nil || !access.Restricted {
return nil
}
if taskKey != "" && access.TaskKeyAllowed(taskKey) {
return nil
}
if access.StageKeyAllowed(access.ActiveStageKey) {
return nil
}
return fmt.Errorf("%w: task %q", bizdev.ErrStageAccessDenied, taskKey)
}
func (g *bizdevStageGate) AllowProjectContextRead(ctx context.Context, projectID, userID int64) error {
access, err := g.access(ctx, projectID, userID)
if err != nil {
return err
}
if access == nil || !access.Restricted {
return nil
}
if len(access.AllowedStageKeys) == 0 {
return bizdev.ErrStageAccessDenied
}
return nil
}
func (g *bizdevStageGate) FilterProjectContext(ctx context.Context, projectID, userID int64, pCtx *ctxmgr.ProjectCollectiveContext) *ctxmgr.ProjectCollectiveContext {
if pCtx == nil {
return pCtx
}
access, err := g.access(ctx, projectID, userID)
if err != nil || access == nil || !access.Restricted {
return pCtx
}
out := *pCtx
out.AllActivities = filterActivitiesByStage(access, pCtx.AllActivities)
out.MilestoneDecisions = filterActivitiesByStage(access, pCtx.MilestoneDecisions)
return &out
}
func filterActivitiesByStage(access *projectinstance.StageAccess, items []ctxmgr.ActivityItem) []ctxmgr.ActivityItem {
if access == nil || !access.Restricted {
return items
}
out := make([]ctxmgr.ActivityItem, 0, len(items))
for _, it := range items {
stageKey := ""
if it.Detail != nil {
if sk, ok := it.Detail["stage_key"].(string); ok {
stageKey = sk
}
}
if stageKey == "" || access.StageKeyAllowed(stageKey) {
out = append(out, it)
}
}
return out
}
func (g *bizdevStageGate) FilterPriorDocs(ctx context.Context, projectID, userID int64, docs []squad.PriorDocItem) []squad.PriorDocItem {
if len(docs) == 0 {
return docs
}
access, err := g.access(ctx, projectID, userID)
if err != nil || access == nil || !access.Restricted {
return docs
}
out := make([]squad.PriorDocItem, 0, len(docs))
for _, d := range docs {
if d.DocKey != "" && access.TaskKeyAllowed(d.DocKey) {
out = append(out, d)
continue
}
if d.Stage != "" && access.StageKeyAllowed(d.Stage) {
out = append(out, d)
}
}
return out
}
internal/server/server.go
+3
View File
@@ -102,6 +102,9 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
notifySvc: notification.NewService(pool),
resolver: workspace.NewResolver(cfg.DataDir, pool),
}
if instSvc != nil && instSvc.Available() {
svc.SetStageAccessGate(newBizdevStageGate(instSvc, pwSvc))
}
s.registerWorkflowApprovalHook()
s.startNotificationReminderLoop()
s.initTaskRunExecutors()