package intentflow_test import ( "context" "errors" "os" "testing" "time" "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/app/uow" "chat-agent-system/internal/db" "github.com/jackc/pgx/v5/pgxpool" ) func init() { if _, err := os.Stat("deploy/postgres/goose"); err == nil { return } for _, cand := range []string{"../../..", "../../../.."} { if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() { _ = os.Chdir(cand) return } } } func TestPlansPG_CreateClaimAndFence(t *testing.T) { if testing.Short() { t.Skip("intentflow PG is not part of -short") } dsn := db.SkipIfNoTestDatabaseURL(t) db.RefuseSharedEnvDSN(t, dsn) ctx := context.Background() pool, err := pgxpool.New(ctx, dsn) if err != nil { t.Fatal(err) } t.Cleanup(pool.Close) if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil { t.Fatalf("migrate: %v", err) } now := time.Date(2026, 9, 15, 15, 0, 0, 0, time.UTC) clock := now owner := intentflow.NewFakeActionOwner() plans := intentflow.NewPlans(uow.NewPG(pool), intentflow.NewRepository(), intentflow.Options{ Owner: owner, Now: func() time.Time { return clock }, Lease: 5 * time.Second, }) prin := intentflow.Principal{UserID: 9, Environment: "ci", ProjectID: 1001, SessionID: "sess-pg-u3"} created, err := plans.CreatePlan(ctx, intentflow.CreateCommand{ Principal: prin, ClientTurnID: "pg-turn-1", RequestDigest: "pg-d1", WritePlan: true, Steps: []intentflow.StepDraft{ {ActionKey: "regulation.query", ActionDigest: "q"}, {ActionKey: "task.run", ActionDigest: "w"}, }, }) if err != nil { t.Fatal(err) } t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM intent_plans WHERE plan_id=$1`, created.PlanID) }) again, err := plans.CreatePlan(ctx, intentflow.CreateCommand{ Principal: prin, ClientTurnID: "pg-turn-1", RequestDigest: "pg-d1", WritePlan: true, Steps: []intentflow.StepDraft{ {ActionKey: "regulation.query", ActionDigest: "q"}, {ActionKey: "task.run", ActionDigest: "w"}, }, }) if err != nil || again.PlanID != created.PlanID { t.Fatalf("idempotent create: %v %+v", err, again) } confirmed, err := plans.DecidePlan(ctx, intentflow.DecisionCommand{ Principal: prin, PlanID: created.PlanID, ExpectedRevision: created.Revision, DecisionNonce: "pg-n", DecisionDigest: "pg-dec", Kind: intentflow.DecisionConfirm, }) if err != nil { t.Fatal(err) } a, err := plans.ClaimStep(ctx, "pg-a", created.PlanID, confirmed.Steps[0].StepID) if err != nil || a == nil { t.Fatalf("claim a: %v", err) } b, err := plans.ClaimStep(ctx, "pg-b", created.PlanID, confirmed.Steps[0].StepID) if err != nil { t.Fatal(err) } if b != nil { t.Fatal("second worker must not own a live lease") } clock = clock.Add(time.Minute) c, err := plans.ClaimStep(ctx, "pg-c", created.PlanID, confirmed.Steps[0].StepID) if err != nil || c == nil { t.Fatalf("takeover: %v", err) } if _, err := plans.ExecuteClaimed(ctx, a); !errors.Is(err, intentflow.ErrPlanActionVersionConflict) { t.Fatalf("old fence: %v", err) } if _, err := plans.ExecuteClaimed(ctx, c); err != nil { t.Fatal(err) } other := prin other.Environment = "dev" if _, err := plans.GetPlan(ctx, other, created.PlanID); !errors.Is(err, intentflow.ErrPlanNotFoundOrHidden) { t.Fatalf("env isolation: %v", err) } }