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

feat(intentflow): re-authorize steps before confirm and execute

Browse Source 
Confirmation stays a command receipt; FakeActionOwner now fail-closes on revoke and resource version drift.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-16 01:01:39 +00:00
co-authored by Cursor
parent f35d265721
commit 08aa56b16c
5 changed files with 323 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.doc/system-design/module-boundaries.yaml
+1
View File
@@ -164,6 +164,7 @@ imports:
- app
- app/chatflow
- app/docflow
- app/intentflow
- app/stageflow
- app/uow
- auth
internal/app/app.go
+1
View File
@@ -193,6 +193,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
rt.Docflow.UoW = uow.NewPG(pool)
rt.Docflow.Events = events.PGSink{}
rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{
Owner: intentflow.NewFakeActionOwner(),
Events: events.PGSink{},
})
}
internal/app/intentflow/authorize_test.go
+126
View File
@@ -0,0 +1,126 @@
package intentflow_test
import (
"context"
"errors"
"testing"
"chat-agent-system/internal/app/intentflow"
)
func TestConfirmDeniedDoesNotMutatePlan(t *testing.T) {
p, _, owner := testPlans()
ctx := context.Background()
created, err := p.CreatePlan(ctx, draft("digest-a"))
if err != nil {
t.Fatal(err)
}
owner.Denied = true
_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-deny", DecisionDigest: "d-deny", Kind: intentflow.DecisionConfirm,
})
if !errors.Is(err, intentflow.ErrPlanActionForbidden) {
t.Fatalf("want forbidden, got %v", err)
}
got, err := p.GetPlan(ctx, principal(), created.PlanID)
if err != nil {
t.Fatal(err)
}
if got.Revision != created.Revision || got.Status != intentflow.PlanAwaitingConfirmation {
t.Fatalf("confirm must be side-effect free on deny: %+v", got)
}
}
func TestResourceVersionConflictOnConfirm(t *testing.T) {
p, _, owner := testPlans()
ctx := context.Background()
cmd := draft("digest-ver")
cmd.Steps[1].ExpectedResourceVersion = "v1"
created, err := p.CreatePlan(ctx, cmd)
if err != nil {
t.Fatal(err)
}
owner.RequiredVersion = map[string]string{"task.run": "v2"}
_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-ver", DecisionDigest: "d-ver", Kind: intentflow.DecisionConfirm,
})
if !errors.Is(err, intentflow.ErrPlanActionVersionConflict) {
t.Fatalf("want version conflict, got %v", err)
}
}
func TestRevokeBetweenConfirmAndExecute(t *testing.T) {
p, _, owner := testPlans()
ctx := context.Background()
created, err := p.CreatePlan(ctx, draft("digest-rev"))
if err != nil {
t.Fatal(err)
}
confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-ok", DecisionDigest: "d-ok", Kind: intentflow.DecisionConfirm,
})
if err != nil {
t.Fatal(err)
}
owner.Denied = true
stepID := confirmed.Steps[0].StepID
claimed, err := p.ClaimStep(ctx, "w1", confirmed.PlanID, stepID)
if err != nil {
t.Fatal(err)
}
if _, err := p.ExecuteClaimed(ctx, claimed); !errors.Is(err, intentflow.ErrPlanActionForbidden) {
t.Fatalf("revoked execute must fail closed, got %v", err)
}
}
func TestEditBumpsRevisionAndInvalidatesOldCard(t *testing.T) {
p, _, _ := testPlans()
ctx := context.Background()
created, err := p.CreatePlan(ctx, draft("digest-edit"))
if err != nil {
t.Fatal(err)
}
edited, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-edit", DecisionDigest: "d-edit", Kind: intentflow.DecisionEdit,
})
if err != nil {
t.Fatal(err)
}
if edited.Revision <= created.Revision {
t.Fatalf("edit must bump revision, got %d", edited.Revision)
}
if edited.Status != intentflow.PlanAwaitingConfirmation {
t.Fatalf("edit stays awaiting, got %s", edited.Status)
}
_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-old", DecisionDigest: "d-old", Kind: intentflow.DecisionConfirm,
})
if !errors.Is(err, intentflow.ErrPlanStaleRevision) {
t.Fatalf("old card must be stale, got %v", err)
}
}
func TestIllegalPartialConfirmMissingDependency(t *testing.T) {
p, _, _ := testPlans()
ctx := context.Background()
cmd := draft("digest-dep")
cmd.Steps[1].DependsOn = []string{"stp_missing"}
created, err := p.CreatePlan(ctx, cmd)
if err != nil {
t.Fatal(err)
}
writeID := created.Steps[1].StepID
_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
DecisionNonce: "n-part", DecisionDigest: "d-part", Kind: intentflow.DecisionConfirm,
StepIDs: []string{writeID},
})
if !errors.Is(err, intentflow.ErrPlanDependencyInvalid) {
t.Fatalf("partial without closure must be invalid, got %v", err)
}
}
internal/app/intentflow/owner.go
+32 -5
View File
@@ -20,11 +20,20 @@ type FenceObserver interface {
// FakeActionOwner is the U3 test double. It records executions and receipts
// without touching business tables.
type FakeActionOwner struct {
mu sync.Mutex
execCount map[string]int
receipts map[string]Receipt
fences map[string]int64
FailWrite bool
mu sync.Mutex
execCount map[string]int
receipts map[string]Receipt
fences map[string]int64
FailWrite bool
Denied bool
DeniedKeys map[string]bool
RequiredVersion map[string]string
}
// StepAuthorizer re-checks the current principal before confirm and execute.
// Confirmation is not authorization.
type StepAuthorizer interface {
Authorize(ctx context.Context, exec StepExec) error
}
func NewFakeActionOwner() *FakeActionOwner {
@@ -44,10 +53,28 @@ func (f *FakeActionOwner) ObserveFence(stepID string, token int64) {
f.fences[stepID] = token
}
func (f *FakeActionOwner) Authorize(ctx context.Context, exec StepExec) error {
if f == nil {
return ErrPlanActionForbidden
}
f.mu.Lock()
defer f.mu.Unlock()
if f.Denied || f.DeniedKeys[exec.ActionKey] {
return ErrPlanActionForbidden
}
if want, ok := f.RequiredVersion[exec.ActionKey]; ok && exec.ExpectedResourceVersion != "" && exec.ExpectedResourceVersion != want {
return ErrPlanActionVersionConflict
}
return nil
}
func (f *FakeActionOwner) Execute(ctx context.Context, exec StepExec) (Receipt, error) {
if f == nil {
return Receipt{}, ErrPlanActionForbidden
}
if err := f.Authorize(ctx, exec); err != nil {
return Receipt{}, err
}
f.mu.Lock()
defer f.mu.Unlock()
if cur, ok := f.fences[exec.StepID]; ok && exec.FencingToken != cur {
internal/app/intentflow/plans.go
+163 -12
View File
@@ -22,6 +22,7 @@ type Plans struct {
uow uow.UnitOfWork
store Store
owner ActionOwner
authz StepAuthorizer
events EventSink
now func() time.Time
newID func(prefix string) string
@@ -30,12 +31,13 @@ type Plans struct {
}
type Options struct {
Owner ActionOwner
Events EventSink
Now func() time.Time
NewID func(prefix string) string
Lease time.Duration
PlanTTL time.Duration
Owner ActionOwner
Authorizer StepAuthorizer
Events EventSink
Now func() time.Time
NewID func(prefix string) string
Lease time.Duration
PlanTTL time.Duration
}
func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans {
@@ -55,7 +57,26 @@ func NewPlans(unit uow.UnitOfWork, store Store, opt Options) *Plans {
if ttl <= 0 {
ttl = defaultPlanTTL
}
return &Plans{uow: unit, store: store, owner: opt.Owner, events: opt.Events, now: now, newID: newID, lease: lease, ttl: ttl}
authz := opt.Authorizer
if authz == nil {
if a, ok := opt.Owner.(StepAuthorizer); ok {
authz = a
}
}
return &Plans{uow: unit, store: store, owner: opt.Owner, authz: authz, events: opt.Events, now: now, newID: newID, lease: lease, ttl: ttl}
}
func (p *Plans) stepAuthorizer() StepAuthorizer {
if p == nil {
return nil
}
if p.authz != nil {
return p.authz
}
if a, ok := p.owner.(StepAuthorizer); ok {
return a
}
return nil
}
func randomID(prefix string) string {
@@ -296,6 +317,9 @@ func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView,
if in.ExpectedRevision != plan.Revision {
return ErrPlanStaleRevision
}
if err := p.authorizeDecision(ctx, plan, in); err != nil {
return err
}
dec := &Decision{
DecisionID: p.newID("dec_"),
PlanID: plan.PlanID,
@@ -329,8 +353,12 @@ func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView,
from := []PlanStatus{plan.Status}
var to PlanStatus
switch in.Kind {
case DecisionConfirm:
if plan.Status != PlanAwaitingConfirmation && plan.Status != PlanAwaitingClarification && plan.Status != PlanProposed {
case DecisionConfirm, DecisionConfirmRetry, DecisionContinue:
if in.Kind == DecisionConfirm {
if plan.Status != PlanAwaitingConfirmation && plan.Status != PlanAwaitingClarification && plan.Status != PlanProposed {
return ErrPlanStaleRevision
}
} else if plan.Status != PlanPartiallyCompleted && plan.Status != PlanExecuting {
return ErrPlanStaleRevision
}
if err := applyConfirm(plan, in.StepIDs, in.ExcludeUnselected, now); err != nil {
@@ -343,6 +371,22 @@ func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView,
}
cancelUnstarted(plan, now)
to = PlanCancelled
case DecisionExclude:
if !isWaitingStatus(plan.Status) && plan.Status != PlanPartiallyCompleted {
return ErrPlanStaleRevision
}
if err := applyExclude(plan, in.StepIDs, now); err != nil {
return err
}
to = plan.Status
if allStepsTerminal(plan) {
to = PlanCancelled
}
case DecisionEdit:
if !isWaitingStatus(plan.Status) {
return ErrPlanStaleRevision
}
to = plan.Status
default:
return ErrPlanDependencyInvalid
}
@@ -364,6 +408,106 @@ func (p *Plans) DecidePlan(ctx context.Context, in DecisionCommand) (*PlanView,
return view, err
}
func (p *Plans) authorizeDecision(ctx context.Context, plan *Plan, in DecisionCommand) error {
switch in.Kind {
case DecisionConfirm, DecisionConfirmRetry, DecisionContinue:
default:
return nil
}
auth := p.stepAuthorizer()
if auth == nil {
return nil
}
ids, err := selectedStepIDs(plan, in.StepIDs)
if err != nil {
return err
}
byID := map[string]Step{}
for _, s := range plan.Steps {
byID[s.StepID] = s
}
for _, id := range ids {
s := byID[id]
if err := auth.Authorize(ctx, StepExec{
Principal: in.Principal,
PlanID: plan.PlanID,
StepID: s.StepID,
ActionKey: s.ActionKey,
ActionDigest: s.ActionDigest,
ExpectedResourceVersion: s.ExpectedResourceVersion,
}); err != nil {
return err
}
}
return nil
}
func selectedStepIDs(plan *Plan, selected []string) ([]string, error) {
byID := map[string]*Step{}
for i := range plan.Steps {
byID[plan.Steps[i].StepID] = &plan.Steps[i]
}
if len(selected) == 0 {
ids := make([]string, 0, len(plan.Steps))
for i := range plan.Steps {
ids = append(ids, plan.Steps[i].StepID)
}
return ids, nil
}
chosen := map[string]bool{}
for _, id := range selected {
if _, ok := byID[id]; !ok {
return nil, ErrPlanDependencyInvalid
}
chosen[id] = true
}
for id := range chosen {
for _, dep := range byID[id].DependsOn {
depStep, ok := byID[dep]
if !ok || (!chosen[dep] && depStep.Status != StepSucceeded && depStep.Status != StepSkipped) {
return nil, ErrPlanDependencyInvalid
}
}
}
ids := make([]string, 0, len(chosen))
for id := range chosen {
ids = append(ids, id)
}
return ids, nil
}
func applyExclude(plan *Plan, stepIDs []string, now time.Time) error {
if len(stepIDs) == 0 {
return ErrPlanDependencyInvalid
}
byID := map[string]*Step{}
for i := range plan.Steps {
byID[plan.Steps[i].StepID] = &plan.Steps[i]
}
for _, id := range stepIDs {
s, ok := byID[id]
if !ok {
return ErrPlanDependencyInvalid
}
if s.Status == StepPending || s.Status == StepEligible {
s.Status = StepCancelled
s.UpdatedAt = now
}
}
return nil
}
func allStepsTerminal(plan *Plan) bool {
for _, s := range plan.Steps {
switch s.Status {
case StepSucceeded, StepFailed, StepDenied, StepSkipped, StepCancelled:
default:
return false
}
}
return true
}
func applyConfirm(plan *Plan, selected []string, excludeUnselected bool, now time.Time) error {
byID := map[string]*Step{}
for i := range plan.Steps {
@@ -383,7 +527,8 @@ func applyConfirm(plan *Plan, selected []string, excludeUnselected bool, now tim
}
for id := range chosen {
for _, dep := range byID[id].DependsOn {
if !chosen[dep] && byID[dep].Status != StepSucceeded && byID[dep].Status != StepSkipped {
depStep, ok := byID[dep]
if !ok || (!chosen[dep] && depStep.Status != StepSucceeded && depStep.Status != StepSkipped) {
return ErrPlanDependencyInvalid
}
}
@@ -462,7 +607,7 @@ func (p *Plans) ExecuteClaimed(ctx context.Context, claimed *ClaimedStep) (*Plan
if p.owner == nil {
return nil, ErrPlanActionForbidden
}
rec, err := p.owner.Execute(ctx, StepExec{
exec := StepExec{
Principal: claimed.Plan.Principal,
PlanID: claimed.Plan.PlanID,
StepID: claimed.Step.StepID,
@@ -471,7 +616,13 @@ func (p *Plans) ExecuteClaimed(ctx context.Context, claimed *ClaimedStep) (*Plan
ActionDigest: claimed.Step.ActionDigest,
ExpectedResourceVersion: claimed.Step.ExpectedResourceVersion,
FencingToken: claimed.FencingToken,
})
}
if auth := p.stepAuthorizer(); auth != nil {
if err := auth.Authorize(ctx, exec); err != nil {
return nil, err
}
}
rec, err := p.owner.Execute(ctx, exec)
if err != nil {
return nil, err
}