weichen
6a3a0dc201
- FIND-SEC-01: GetUserContext requires session ownership (handler 403 + SQL/memory owner filter); in-memory HTTP + isolated-PG regressions - FIND-ARCH-01: archtest fails closed on untracked table writes (+noise exemption) and registers 50 existing writer tables in table_owners - FIND-SEC-02: Authorizer without pool denies by default; explicit NewInMemoryAuthorizer seam opt-in via ALLOW_IN_MEMORY at composition - FIND-SEC-03: swboot commands run argv-exec (no sh -c) with exact basename allowlist match; prefix-spoof and no-shell-eval tests - FIND-FLOW-01: confirm rejects validated_only plans; ExecuteClaimed hard-refuses validated_only steps (new error codes + HTTP mapping) - FIND-QUAL-01: intentflow.ValidateFlagConsistency fail-fast at Assemble
395 lines
12 KiB
Go
395 lines
12 KiB
Go
package intentflow_test
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"sync"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
|
||
"chat-agent-system/internal/app/intentflow"
|
||
)
|
||
|
||
func testClock() (now *time.Time, opts intentflow.Options, owner *intentflow.FakeActionOwner) {
|
||
t0 := time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC)
|
||
now = &t0
|
||
owner = intentflow.NewFakeActionOwner()
|
||
opts = intentflow.Options{
|
||
Owner: owner,
|
||
Now: func() time.Time { return now.UTC() },
|
||
Lease: 10 * time.Second,
|
||
PlanTTL: 30 * time.Minute,
|
||
}
|
||
return now, opts, owner
|
||
}
|
||
|
||
func testPlans() (*intentflow.Plans, *time.Time, *intentflow.FakeActionOwner) {
|
||
now, opts, owner := testClock()
|
||
return intentflow.NewPlans(&intentflow.SerialUoW{}, intentflow.NewMemoryStore(), opts), now, owner
|
||
}
|
||
|
||
func principal() intentflow.Principal {
|
||
return intentflow.Principal{UserID: 7, Environment: "dev", ProjectID: 42, SessionID: "sess-u3"}
|
||
}
|
||
|
||
func draft(digest string) intentflow.CreateCommand {
|
||
return intentflow.CreateCommand{
|
||
Principal: principal(),
|
||
ClientTurnID: "turn-1",
|
||
RequestDigest: digest,
|
||
WritePlan: true,
|
||
SafeSummary: "query then generate",
|
||
// 这些用例验证 live 确认/派发机制；CreateCommand 缺省 mode 是
|
||
// validated_only（安全默认，不可 Confirm），必须显式声明 live。
|
||
ExecutionMode: intentflow.ModeLive,
|
||
Steps: []intentflow.StepDraft{
|
||
{ActionKey: "regulation.query", ActionDigest: "q1"},
|
||
{ActionKey: "task.run", ActionDigest: "w1"},
|
||
},
|
||
}
|
||
}
|
||
|
||
func TestHandleTurnLegacyDoesNotPersist(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
ctx := context.Background()
|
||
res, err := p.HandleTurn(ctx, intentflow.TurnInput{Principal: principal(), Text: "生成报价单", ClientTurnID: "turn-1", RequestDigest: "d1"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if res == nil || res.Plan != nil {
|
||
t.Fatalf("legacy HandleTurn must not create a plan: %+v", res)
|
||
}
|
||
active, err := p.GetActivePlan(ctx, principal())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if active != nil {
|
||
t.Fatal("no plan should be visible")
|
||
}
|
||
}
|
||
|
||
func TestCreateIdempotentSameDigest(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
ctx := context.Background()
|
||
a, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
b, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if a.PlanID != b.PlanID {
|
||
t.Fatalf("same turn/digest must reuse plan %s vs %s", a.PlanID, b.PlanID)
|
||
}
|
||
other := draft("digest-b")
|
||
if _, err := p.CreatePlan(ctx, other); !errors.Is(err, intentflow.ErrPlanDecisionConflict) {
|
||
t.Fatalf("different digest must conflict, got %v", err)
|
||
}
|
||
}
|
||
|
||
func TestCreateRejectsSecondActiveWritePlan(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
ctx := context.Background()
|
||
first, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
second := draft("digest-x")
|
||
second.ClientTurnID = "turn-2"
|
||
got, err := p.CreatePlan(ctx, second)
|
||
if !errors.Is(err, intentflow.ErrPlanDecisionConflict) {
|
||
t.Fatalf("want conflict, got %v", err)
|
||
}
|
||
if got == nil || got.PlanID != first.PlanID {
|
||
t.Fatalf("must return the active write plan, got %+v", got)
|
||
}
|
||
}
|
||
|
||
func TestDecisionNonceSameAndDifferentDigest(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
ctx := context.Background()
|
||
created, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmd := intentflow.DecisionCommand{
|
||
Principal: principal(),
|
||
PlanID: created.PlanID,
|
||
ExpectedRevision: created.Revision,
|
||
DecisionNonce: "nonce-1",
|
||
DecisionDigest: "dec-1",
|
||
Kind: intentflow.DecisionConfirm,
|
||
}
|
||
first, err := p.DecidePlan(ctx, cmd)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
again, err := p.DecidePlan(ctx, cmd)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if first.Revision != again.Revision {
|
||
t.Fatal("same nonce/digest must return the same receipt")
|
||
}
|
||
cmd.DecisionDigest = "dec-other"
|
||
if _, err := p.DecidePlan(ctx, cmd); !errors.Is(err, intentflow.ErrPlanDecisionConflict) {
|
||
t.Fatalf("different digest must conflict, got %v", err)
|
||
}
|
||
}
|
||
|
||
func TestConfirmCancelRaceOneWinner(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
ctx := context.Background()
|
||
created, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var okConfirm, okCancel atomic.Int32
|
||
var wg sync.WaitGroup
|
||
wg.Add(2)
|
||
go func() {
|
||
defer wg.Done()
|
||
_, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
|
||
DecisionNonce: "n-confirm", DecisionDigest: "d-confirm", Kind: intentflow.DecisionConfirm,
|
||
})
|
||
if err == nil {
|
||
okConfirm.Add(1)
|
||
}
|
||
}()
|
||
go func() {
|
||
defer wg.Done()
|
||
_, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
|
||
DecisionNonce: "n-cancel", DecisionDigest: "d-cancel", Kind: intentflow.DecisionCancel,
|
||
})
|
||
if err == nil {
|
||
okCancel.Add(1)
|
||
}
|
||
}()
|
||
wg.Wait()
|
||
if okConfirm.Load()+okCancel.Load() != 1 {
|
||
t.Fatalf("exactly one CAS winner, confirm=%d cancel=%d", okConfirm.Load(), okCancel.Load())
|
||
}
|
||
}
|
||
|
||
func TestTwoWorkersOneOwnerAndLeaseTakeover(t *testing.T) {
|
||
p, clock, owner := testPlans()
|
||
ctx := context.Background()
|
||
created, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
|
||
DecisionNonce: "n-c", DecisionDigest: "d-c", Kind: intentflow.DecisionConfirm,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
stepID := confirmed.Steps[0].StepID
|
||
var gotA, gotB atomic.Int32
|
||
var first *intentflow.ClaimedStep
|
||
var mu sync.Mutex
|
||
var wg sync.WaitGroup
|
||
wg.Add(2)
|
||
claim := func(worker string, slot *atomic.Int32) {
|
||
defer wg.Done()
|
||
c, err := p.ClaimStep(ctx, worker, created.PlanID, stepID)
|
||
if err != nil {
|
||
t.Errorf("claim %s: %v", worker, err)
|
||
return
|
||
}
|
||
if c != nil {
|
||
slot.Add(1)
|
||
mu.Lock()
|
||
if first == nil {
|
||
first = c
|
||
}
|
||
mu.Unlock()
|
||
}
|
||
}
|
||
go claim("worker-a", &gotA)
|
||
go claim("worker-b", &gotB)
|
||
wg.Wait()
|
||
if gotA.Load()+gotB.Load() != 1 {
|
||
t.Fatalf("two workers must not both own, a=%d b=%d", gotA.Load(), gotB.Load())
|
||
}
|
||
stale := first
|
||
*clock = clock.Add(time.Minute)
|
||
takeover, err := p.ClaimStep(ctx, "worker-c", created.PlanID, stepID)
|
||
if err != nil || takeover == nil {
|
||
t.Fatalf("expired lease must be taken over: %v %+v", err, takeover)
|
||
}
|
||
if takeover.FencingToken <= stale.FencingToken {
|
||
t.Fatalf("fencing must be monotonic: old=%d new=%d", stale.FencingToken, takeover.FencingToken)
|
||
}
|
||
if _, err := p.ExecuteClaimed(ctx, stale); !errors.Is(err, intentflow.ErrPlanActionVersionConflict) {
|
||
t.Fatalf("old fence must be rejected by owner, got %v", err)
|
||
}
|
||
if _, err := p.ExecuteClaimed(ctx, takeover); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if owner.ExecCount(takeover.IdempotencyKey) != 1 {
|
||
t.Fatalf("execute count = %d", owner.ExecCount(takeover.IdempotencyKey))
|
||
}
|
||
}
|
||
|
||
func TestReceiptWritebackFailureDoesNotRerun(t *testing.T) {
|
||
p, _, owner := testPlans()
|
||
ctx := context.Background()
|
||
created, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
confirmed, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: created.Revision,
|
||
DecisionNonce: "n-c", DecisionDigest: "d-c", Kind: intentflow.DecisionConfirm,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
claimed, err := p.ClaimStep(ctx, "w1", created.PlanID, confirmed.Steps[0].StepID)
|
||
if err != nil || claimed == nil {
|
||
t.Fatalf("claim: %v %+v", err, claimed)
|
||
}
|
||
if _, err := owner.Execute(ctx, intentflow.StepExec{
|
||
Principal: principal(), PlanID: claimed.Plan.PlanID, StepID: claimed.Step.StepID,
|
||
ActionKey: claimed.Step.ActionKey, IdempotencyKey: claimed.IdempotencyKey,
|
||
ActionDigest: claimed.Step.ActionDigest, FencingToken: claimed.FencingToken,
|
||
}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := p.RecoverReceipt(ctx, claimed); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if owner.ExecCount(claimed.IdempotencyKey) != 1 {
|
||
t.Fatalf("recover must not re-execute, count=%d", owner.ExecCount(claimed.IdempotencyKey))
|
||
}
|
||
got, err := p.GetPlan(ctx, principal(), created.PlanID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got.Steps[0].Status != intentflow.StepSucceeded {
|
||
t.Fatalf("step status %s", got.Steps[0].Status)
|
||
}
|
||
}
|
||
|
||
func TestExpireAndEnvironmentIsolationAndRetention(t *testing.T) {
|
||
p, clock, _ := testPlans()
|
||
ctx := context.Background()
|
||
created, err := p.CreatePlan(ctx, draft("digest-a"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
other := principal()
|
||
other.Environment = "test"
|
||
if _, err := p.GetPlan(ctx, other, created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) {
|
||
t.Fatalf("cross-env must hide, got %v", err)
|
||
}
|
||
*clock = clock.Add(time.Hour)
|
||
n, err := p.ExpireDue(ctx)
|
||
if err != nil || n != 1 {
|
||
t.Fatalf("expire n=%d err=%v", n, err)
|
||
}
|
||
if _, err := p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(), PlanID: created.PlanID, ExpectedRevision: 1,
|
||
DecisionNonce: "late", DecisionDigest: "late", Kind: intentflow.DecisionConfirm,
|
||
}); !errors.Is(err, intentflow.ErrPlanExpired) && !errors.Is(err, intentflow.ErrPlanStaleRevision) {
|
||
got, _ := p.GetPlan(ctx, principal(), created.PlanID)
|
||
if got == nil || got.Status != intentflow.PlanExpired {
|
||
t.Fatalf("expired plan confirm: %v view=%+v", err, got)
|
||
}
|
||
}
|
||
purged, err := p.PurgeClosed(ctx, clock.Add(time.Second))
|
||
if err != nil || purged != 1 {
|
||
t.Fatalf("purge n=%d err=%v", purged, err)
|
||
}
|
||
if _, err := p.GetPlan(ctx, principal(), created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) {
|
||
t.Fatalf("purged plan must be hidden, got %v", err)
|
||
}
|
||
}
|
||
|
||
func TestUnauthenticatedRejected(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
_, err := p.CreatePlan(context.Background(), intentflow.CreateCommand{
|
||
Principal: intentflow.Principal{}, ClientTurnID: "t", RequestDigest: "d",
|
||
Steps: []intentflow.StepDraft{{ActionKey: "regulation.query"}},
|
||
})
|
||
if !errors.Is(err, intentflow.ErrUnauthenticated) {
|
||
t.Fatalf("got %v", err)
|
||
}
|
||
}
|
||
|
||
// TestConfirmRejectedForValidatedOnlyPlan FIND-FLOW-01 回归：
|
||
// validated_only（shadow）的 proposed 计划不允许经 Confirm 进入 executing。
|
||
func TestConfirmRejectedForValidatedOnlyPlan(t *testing.T) {
|
||
p, _, _ := testPlans()
|
||
ctx := context.Background()
|
||
cmd := draft("digest-vo")
|
||
cmd.Status = intentflow.PlanProposed
|
||
cmd.ExecutionMode = intentflow.ModeValidatedOnly
|
||
created, err := p.CreatePlan(ctx, cmd)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(),
|
||
PlanID: created.PlanID,
|
||
ExpectedRevision: created.Revision,
|
||
DecisionNonce: "n-vo",
|
||
DecisionDigest: "d-vo",
|
||
Kind: intentflow.DecisionConfirm,
|
||
})
|
||
if !errors.Is(err, intentflow.ErrPlanModeNotConfirmable) {
|
||
t.Fatalf("confirm on validated_only plan must be rejected with ErrPlanModeNotConfirmable, got %v", err)
|
||
}
|
||
}
|
||
|
||
// TestExecuteClaimedRefusesValidatedOnlyStep FIND-FLOW-01 回归：
|
||
// 执行器硬读 ExecutionMode——validated_only 计划即便被误推进到 executing
|
||
// 并经 DispatchEligible 派发，也拒绝真实执行（同 http-confirm 派发入口）。
|
||
func TestExecuteClaimedRefusesValidatedOnlyStep(t *testing.T) {
|
||
p, _, owner := testPlans()
|
||
ctx := context.Background()
|
||
cmd := draft("digest-vo-exec")
|
||
// 直接构造「上游状态机被误推进」的形状：executing + validated_only。
|
||
cmd.Status = intentflow.PlanExecuting
|
||
cmd.ExecutionMode = intentflow.ModeValidatedOnly
|
||
created, err := p.CreatePlan(ctx, cmd)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// DecisionContinue 允许 executing 计划推进（applyConfirm 将步骤置为
|
||
// eligible），模拟状态机被误推进后的派发路径。
|
||
_, err = p.DecidePlan(ctx, intentflow.DecisionCommand{
|
||
Principal: principal(),
|
||
PlanID: created.PlanID,
|
||
ExpectedRevision: created.Revision,
|
||
DecisionNonce: "n-vo-exec",
|
||
DecisionDigest: "d-vo-exec",
|
||
Kind: intentflow.DecisionContinue,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
_, err = p.DispatchEligible(ctx, "worker-vo", created.PlanID)
|
||
if !errors.Is(err, intentflow.ErrPlanModeNotExecutable) {
|
||
t.Fatalf("dispatch of validated_only plan must hard-refuse with ErrPlanModeNotExecutable, got %v", err)
|
||
}
|
||
plan, err := p.GetPlan(ctx, principal(), created.PlanID)
|
||
if err != nil || plan == nil {
|
||
t.Fatalf("get plan: %v %v", plan, err)
|
||
}
|
||
for _, s := range plan.Steps {
|
||
if s.Status == intentflow.StepSucceeded {
|
||
t.Fatalf("validated_only step must never be written as succeeded: %+v", s)
|
||
}
|
||
if owner.ExecCount(s.StepID) != 0 {
|
||
t.Fatalf("owner must not observe real execution for %s", s.StepID)
|
||
}
|
||
}
|
||
}
|