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) } } }