feat(purchase): per-plan step commitment cap, auto-purchase kill switch and notified veto

internal/api/handler_plans.go

@@ -339,6 +339,10 @@ func mergeUpdatedPlan(existingPlan *config.PurchasePlan, req *PlanRequest, planI
 		return nil, err
 	}
 
+	if req.MaxStepCommitment == nil {
+		plan.MaxStepCommitment = existingPlan.MaxStepCommitment
+	}
+
 	// If no services were created from request, preserve existing services
 	if len(plan.Services) == 0 && len(existingPlan.Services) > 0 {
 		plan.Services = existingPlan.Services

internal/api/handler_plans_step_cap_test.go

@@ -0,0 +1,36 @@
+package api
+
+import (
+	"math"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMergeUpdatedPlan_MaxStepCommitment(t *testing.T) {
+	cap100, cap5 := 100.0, 5.0
+	existing := &config.PurchasePlan{Name: "p", MaxStepCommitment: &cap100,
+		RampSchedule: config.RampSchedule{Type: "immediate", PercentPerStep: 100, TotalSteps: 1}}
+
+	t.Run("omitted on update keeps the current cap", func(t *testing.T) {
+		plan, err := mergeUpdatedPlan(existing, &PlanRequest{Name: "p", RampSchedule: "immediate"}, "id")
+		require.NoError(t, err)
+		require.NotNil(t, plan.MaxStepCommitment)
+		assert.Equal(t, 100.0, *plan.MaxStepCommitment)
+	})
+	t.Run("provided on update replaces it", func(t *testing.T) {
+		plan, err := mergeUpdatedPlan(existing, &PlanRequest{Name: "p", RampSchedule: "immediate", MaxStepCommitment: &cap5}, "id")
+		require.NoError(t, err)
+		assert.Equal(t, 5.0, *plan.MaxStepCommitment)
+	})
+	for name, v := range map[string]float64{"zero": 0, "negative": -1, "NaN": math.NaN(), "Inf": math.Inf(1)} {
+		t.Run("rejects "+name, func(t *testing.T) {
+			bad := v
+			_, err := mergeUpdatedPlan(existing, &PlanRequest{Name: "p", RampSchedule: "immediate", MaxStepCommitment: &bad}, "id")
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), "max_step_commitment")
+		})
+	}
+}

internal/api/handler_purchases.go

@@ -2586,11 +2586,7 @@ func purchaseConstraintSets(recs []config.RecommendationRecord) []auth.Permissio
 // execute path the two cost fields are the store-derived values written by
 // priceRecommendationsFromStore.
 func recTotalCommitment(rec *config.RecommendationRecord) float64 {
-	total := rec.UpfrontCost
-	if rec.Term > 0 && rec.MonthlyCost != nil {
-		total += *rec.MonthlyCost * float64(rec.Term*12)
-	}
-	return total
+	return rec.TotalCommitment()
 }
 
 // requireNonZeroCommitment guards against a purchase batch whose computed

internal/api/handler_purchases_pause_claim_integration_test.go

@@ -67,6 +67,19 @@ type pauseClaimFixture struct {
 	workers   sync.WaitGroup
 }
 
+// pauseClaimCap is a plan cap far above the fixture recommendations.
+var pauseClaimCap = 1e9
+
+// markNotified moves the row to "notified", the only state in which the
+// scheduler and SQS paths will auto-execute a recommendation-bearing plan row
+// (the notification email is the veto window). pause only applies to "pending",
+// so tests call this after pausing and resuming and before the worker runs.
+func (f *pauseClaimFixture) markNotified(t *testing.T) {
+	t.Helper()
+	_, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET status = 'notified' WHERE execution_id = $1`, f.execution.ExecutionID)
+	require.NoError(t, err)
+}
+
 func newPauseClaimFixture(t *testing.T) *pauseClaimFixture {
 	t.Helper()
 	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
@@ -75,7 +88,12 @@ func newPauseClaimFixture(t *testing.T) *pauseClaimFixture {
 	plan, err := f.store.GetPurchasePlan(ctx, f.planID)
 	require.NoError(t, err)
 	plan.AutoPurchase = true
+	plan.MaxStepCommitment = &pauseClaimCap
 	require.NoError(t, f.store.UpdatePurchasePlan(ctx, plan))
+	globalCfg, err := f.store.GetGlobalConfig(ctx)
+	require.NoError(t, err)
+	globalCfg.AutoPurchaseEnabled = true
+	require.NoError(t, f.store.SaveGlobalConfig(ctx, globalCfg))
 	accounts, err := f.store.GetPlanAccounts(ctx, plan.ID)
 	require.NoError(t, err)
 	require.Len(t, accounts, 1)
@@ -187,6 +205,7 @@ func assertPauseConflict(t *testing.T, err error) {
 
 func TestPauseClaim_RunningWorkerCannotBeReclaimed(t *testing.T) {
 	f := newPauseClaimFixture(t)
+	f.markNotified(t)
 	first := f.startWorker()
 	select {
 	case <-f.service.entered:
@@ -222,6 +241,11 @@ func TestPauseClaim_PausedQueueResumesExactlyOnce(t *testing.T) {
 	require.Equal(t, paused, f.state(t))
 	require.NoError(t, f.resume())
 	require.Equal(t, "pending", f.state(t).Status)
+	// A resumed row is pending again, so auto-execution still refuses it until
+	// a notification has gone out.
+	require.Error(t, f.manager.ProcessMessage(f.ctx, f.body))
+	require.Zero(t, f.service.calls.Load())
+	f.markNotified(t)
 	f.service.unblock()
 	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
 	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
@@ -253,22 +277,20 @@ func TestPauseClaim_ConcurrentPauseAndWorker(t *testing.T) {
 	close(start)
 	select {
 	case err := <-paused:
-		if err == nil {
-			f.waitWorker(t, worker)
-			require.Zero(t, f.service.calls.Load())
-			require.Equal(t, "paused", f.state(t).Status)
-			require.NoError(t, f.resume())
-			f.service.unblock()
-			require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
-		} else {
-			assertPauseConflict(t, err)
-			require.Equal(t, "running", f.state(t).Status)
-			require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
-			f.service.unblock()
-			f.waitWorker(t, worker)
-		}
+		require.NoError(t, err, "a pending row is pausable: the worker can never claim it")
 	case <-f.ctx.Done():
 		t.Fatal("pause did not finish")
 	}
+	select {
+	case <-worker: // refused (row still pending) or lost the claim (row paused): both buy nothing
+	case <-f.ctx.Done():
+		t.Fatal("purchase worker did not finish")
+	}
+	require.Zero(t, f.service.calls.Load())
+	require.Equal(t, "paused", f.state(t).Status)
+	require.NoError(t, f.resume())
+	f.markNotified(t)
+	f.service.unblock()
+	require.NoError(t, f.manager.ProcessMessage(f.ctx, f.body))
 	f.assertCompleted(t)
 }

internal/api/openapi.yaml

@@ -3170,6 +3170,14 @@ components:
           type: boolean
         notification_days_before:
           type: integer
+        max_step_commitment:
+          type: number
+          exclusiveMinimum: 0
+          description: >-
+            Ceiling on the total commitment (upfront + monthly * term) one ramp
+            step may auto-purchase. Omit on create for no cap (auto-purchase of
+            steps with recommendations is then refused); omit on update to keep
+            the current cap.
         provider:
           type: string
         service:
@@ -3212,6 +3220,10 @@ components:
           type: boolean
         notification_days_before:
           type: integer
+        max_step_commitment:
+          type: number
+          nullable: true
+          description: Cap on the total commitment one ramp step may auto-purchase; null means auto-purchase of steps with recommendations is refused.
         services:
           type: object
           additionalProperties:

internal/api/types.go

@@ -800,8 +800,12 @@ type PlanRequest struct {
 	NotificationDaysBefore int      `json:"notification_days_before"`
 	CustomStepPercent      int      `json:"custom_step_percent,omitempty"`
 	CustomIntervalDays     int      `json:"custom_interval_days,omitempty"`
-	AutoPurchase           bool     `json:"auto_purchase"`
-	Enabled                bool     `json:"enabled"`
+	// MaxStepCommitment caps the total commitment one ramp step may
+	// auto-purchase. Omitted on create = no cap (auto-purchase of steps with
+	// recommendations stays refused); omitted on update = keep the current cap.
+	MaxStepCommitment *float64 `json:"max_step_commitment,omitempty"`
+	AutoPurchase      bool     `json:"auto_purchase"`
+	Enabled           bool     `json:"enabled"`
 }
 
 // toPurchasePlan converts a PlanRequest to a config.PurchasePlan.
@@ -812,6 +816,7 @@ func (r *PlanRequest) toPurchasePlan() *config.PurchasePlan {
 		Enabled:                r.Enabled,
 		AutoPurchase:           r.AutoPurchase,
 		NotificationDaysBefore: r.NotificationDaysBefore,
+		MaxStepCommitment:      r.MaxStepCommitment,
 		CreatedAt:              now,
 		UpdatedAt:              now,
 	}

internal/config/commitment.go

@@ -0,0 +1,16 @@
+package config
+
+// TotalCommitment returns the recommendation's full committed spend: the
+// upfront cost plus the recurring monthly cost over the term. UpfrontCost and
+// MonthlyCost are totals for the rec's Count, so they sum with no per-instance
+// scaling. Term is in years (converted with *12). A nil MonthlyCost (all-upfront
+// commitments) or a non-positive Term contributes nothing to the recurring leg.
+// Used for spend caps instead of UpfrontCost alone, which is $0 for no-upfront
+// commitments (Azure monthly billing).
+func (r *RecommendationRecord) TotalCommitment() float64 {
+	total := r.UpfrontCost
+	if r.Term > 0 && r.MonthlyCost != nil {
+		total += *r.MonthlyCost * float64(r.Term*12)
+	}
+	return total
+}

internal/config/store_postgres.go

@@ -92,7 +92,8 @@ func getGlobalConfigFrom(ctx context.Context, q globalConfigExecutor) (*GlobalCo
 		       COALESCE(laddering_enabled, false),
 		       COALESCE(ladder_execution_enabled, false),
 		       offering_class,
-		       require_different_approver
+		       require_different_approver,
+		       auto_purchase_enabled
 		FROM global_config
 		WHERE id = 1
 	`
@@ -126,6 +127,7 @@ func getGlobalConfigFrom(ctx context.Context, q globalConfigExecutor) (*GlobalCo
 		&config.LadderExecutionEnabled,
 		&config.OfferingClass,
 		&config.RequireDifferentApprover,
+		&config.AutoPurchaseEnabled,
 	)
 
 	if err != nil {
@@ -259,8 +261,8 @@ func saveGlobalConfigWith(ctx context.Context, q globalConfigExecutor, config *G
 			grace_period_days,
 			recommendations_cache_stale_hours, recommendations_lookback_days,
 			purchase_delay_hours, laddering_enabled, ladder_execution_enabled, offering_class,
-			require_different_approver
-		) VALUES (1, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24)
+			require_different_approver, auto_purchase_enabled
+		) VALUES (1, $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25)
 		ON CONFLICT (id) DO UPDATE SET
 			enabled_providers = $1,
 			notification_email = $2,
@@ -286,6 +288,7 @@ func saveGlobalConfigWith(ctx context.Context, q globalConfigExecutor, config *G
 			ladder_execution_enabled = $22,
 			offering_class = $23,
 			require_different_approver = $24,
+			auto_purchase_enabled = $25,
 			updated_at = NOW()
 	`
 
@@ -354,6 +357,7 @@ func saveGlobalConfigWith(ctx context.Context, q globalConfigExecutor, config *G
 		config.LadderExecutionEnabled,
 		offeringClass,
 		config.RequireDifferentApprover,
+		config.AutoPurchaseEnabled,
 	)
 
 	if err != nil {
@@ -556,8 +560,9 @@ func (s *PostgresStore) CreatePurchasePlan(ctx context.Context, plan *PurchasePl
 		INSERT INTO purchase_plans (
 			id, name, enabled, auto_purchase, notification_days_before,
 			services, ramp_schedule, created_at, updated_at,
-			next_execution_date, last_execution_date, last_notification_sent
-		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+			next_execution_date, last_execution_date, last_notification_sent,
+			max_step_commitment
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
 		RETURNING updated_at
 	`
 
@@ -575,6 +580,7 @@ func (s *PostgresStore) CreatePurchasePlan(ctx context.Context, plan *PurchasePl
 		plan.NextExecutionDate,
 		plan.LastExecutionDate,
 		plan.LastNotificationSent,
+		plan.MaxStepCommitment,
 	).Scan(&updatedAt)
 
 	if err != nil {
@@ -588,7 +594,8 @@ func (s *PostgresStore) CreatePurchasePlan(ctx context.Context, plan *PurchasePl
 const purchasePlanSelectCols = `
 	SELECT id, name, enabled, auto_purchase, notification_days_before,
 	       services, ramp_schedule, created_at, updated_at,
-	       next_execution_date, last_execution_date, last_notification_sent
+	       next_execution_date, last_execution_date, last_notification_sent,
+	       max_step_commitment
 	FROM purchase_plans`
 
 // scanPurchasePlanRow deserialises one purchase_plans row returned by QueryRow
@@ -612,6 +619,7 @@ func scanPurchasePlanRow(row pgx.Row) (*PurchasePlan, error) {
 		&nextExecDate,
 		&lastExecDate,
 		&lastNotifSent,
+		&plan.MaxStepCommitment,
 	)
 	if err != nil {
 		return nil, err
@@ -799,7 +807,8 @@ func (s *PostgresStore) UpdatePurchasePlanTx(ctx context.Context, tx pgx.Tx, pla
 			ramp_schedule = $7,
 			next_execution_date = $9,
 			last_execution_date = $10,
-			last_notification_sent = $11
+			last_notification_sent = $11,
+			max_step_commitment = $12
 		WHERE id = $1 AND updated_at = $8
 		RETURNING updated_at
 	`
@@ -817,6 +826,7 @@ func (s *PostgresStore) UpdatePurchasePlanTx(ctx context.Context, tx pgx.Tx, pla
 		plan.NextExecutionDate,
 		plan.LastExecutionDate,
 		plan.LastNotificationSent,
+		plan.MaxStepCommitment,
 	).Scan(&updatedAt)
 
 	if errors.Is(err, pgx.ErrNoRows) {
@@ -888,6 +898,7 @@ func buildListPlansQuery(accountIDs []string, includeUnassigned bool) (query str
 			SELECT id, name, enabled, auto_purchase, notification_days_before,
 			       services, ramp_schedule, created_at, updated_at,
 			       next_execution_date, last_execution_date, last_notification_sent,
+			       max_step_commitment,
 			       false AS unassigned
 			FROM purchase_plans
 			ORDER BY created_at DESC
@@ -907,6 +918,7 @@ func buildListPlansQuery(accountIDs []string, includeUnassigned bool) (query str
 		SELECT DISTINCT pp.id, pp.name, pp.enabled, pp.auto_purchase, pp.notification_days_before,
 		       pp.services, pp.ramp_schedule, pp.created_at, pp.updated_at,
 		       pp.next_execution_date, pp.last_execution_date, pp.last_notification_sent,
+		       pp.max_step_commitment,
 		       (NOT EXISTS (SELECT 1 FROM plan_accounts WHERE plan_id = pp.id)) AS unassigned
 		FROM purchase_plans pp
 		LEFT JOIN plan_accounts pa ON pa.plan_id = pp.id
@@ -948,6 +960,7 @@ func (s *PostgresStore) ListPurchasePlans(ctx context.Context, filter PurchasePl
 			&nextExecDate,
 			&lastExecDate,
 			&lastNotifSent,
+			&plan.MaxStepCommitment,
 			&plan.Unassigned,
 		)
 		if err != nil {

internal/config/store_postgres_complete_step_test.go

@@ -56,6 +56,7 @@ var purchasePlanRowCols = []string{
 	"id", "name", "enabled", "auto_purchase", "notification_days_before",
 	"services", "ramp_schedule", "created_at", "updated_at",
 	"next_execution_date", "last_execution_date", "last_notification_sent",
+	"max_step_commitment",
 }
 
 // rampStepArg is a pgxmock.Argument that unmarshals the ramp_schedule JSONB
@@ -87,7 +88,7 @@ func (a nullTimeArg) Match(v interface{}) bool {
 	return (tp == nil) == a.wantNil
 }
 
-const purchasePlanUpdateArgs = 11
+const purchasePlanUpdateArgs = 12
 
 // completeStepUpdateArgs pins the completed step, expected read version, and next-date presence.
 func completeStepUpdateArgs(wantStep int, staleUpdatedAt time.Time, wantNextNil bool) []interface{} {
@@ -134,7 +135,7 @@ func rampPlanRows(t *testing.T, planID string, ramp RampSchedule, now, stale tim
 	return pgxmock.NewRows(purchasePlanRowCols).AddRow(
 		planID, "Ramp Plan", true, true, 3,
 		svcJSON, rampJSON, now, stale,
-		nextExec, sql.NullTime{Valid: false}, sql.NullTime{Valid: false},
+		nextExec, sql.NullTime{Valid: false}, sql.NullTime{Valid: false}, (*float64)(nil),
 	)
 }
 

internal/config/store_postgres_coverage_test.go

@@ -569,7 +569,7 @@ func TestSaveGlobalConfig_OfferingClassBindsAt23(t *testing.T) {
 		OfferingClass:       "standard",
 	}
 
-	// Expect exactly 24 args; pgxmock validates arg count and types.
+	// Expect exactly 25 args; pgxmock validates arg count and types.
 	// The 21st arg is laddering_enabled; the 22nd is ladder_execution_enabled;
 	// the 23rd arg must be "standard" (offering_class); the 24th is
 	// require_different_approver (issue #1005).
@@ -601,6 +601,7 @@ func TestSaveGlobalConfig_OfferingClassBindsAt23(t *testing.T) {
 			pgxmock.AnyArg(), // $22 ladder_execution_enabled
 			"standard",       // $23 offering_class -- the field this test guards
 			pgxmock.AnyArg(), // $24 require_different_approver
+			pgxmock.AnyArg(), // $25 auto_purchase_enabled
 		).
 		WillReturnResult(pgxmock.NewResult("INSERT", 1))