Agent-System/internal/app/intentflow/plans_test.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",
		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)
	}
}