fix(purchase): refuse unverified ramp sizing and adoption

internal/config/store_postgres_plan_writes.go

@@ -34,8 +34,8 @@ func (s *PostgresStore) createPurchasePlan(ctx context.Context, executor globalC
 			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
-		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
+			max_step_commitment, ramp_ledger_state
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, 'new')
 		RETURNING updated_at
 	`
 

internal/config/store_postgres_ramp_baseline.go

@@ -21,11 +21,21 @@ type PlanRampQuantity struct {
 // PlanRampBaseline maps RecIdentityKey to the base a ramp plan sizes against.
 type PlanRampBaseline map[string]PlanRampQuantity
 
-// SetPlanRampBaselineTx freezes a plan's ramp baseline. It writes only while
-// the column is NULL and reports whether it did: the baseline is immutable
-// once set. It has its own UPDATE and the plan's full-row UPDATE
-// (UpdatePurchasePlanTx) never names the column, so a plan saved from a stale
-// in-memory struct cannot clear or replace it.
+type rampLedgerState string
+
+const (
+	rampLedgerNew     rampLedgerState = "new"
+	rampLedgerTrusted rampLedgerState = "trusted"
+)
+
+func (s rampLedgerState) allowsSizing(raw []byte, current int) bool {
+	return s == rampLedgerTrusted && raw != nil || s == rampLedgerNew && raw == nil && current == 0
+}
+
+var ErrRampLedgerUntrusted = errors.New("this ramp's purchase history from before the retention ledger cannot be verified, so it cannot size further steps")
+
+// SetPlanRampBaselineTx freezes a baseline only while retention can prove its lineage.
+// The caller holds the plan lock; full-row plan writes cannot promote trust.
 func (s *PostgresStore) SetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, planID string, baseline PlanRampBaseline) (bool, error) {
 	if len(baseline) == 0 {
 		return false, fmt.Errorf("refusing to freeze an empty ramp baseline for plan %s", planID)
@@ -35,24 +45,36 @@ func (s *PostgresStore) SetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, pl
 		return false, fmt.Errorf("failed to marshal ramp baseline of plan %s: %w", planID, err)
 	}
 	tag, err := tx.Exec(ctx,
-		`UPDATE purchase_plans SET ramp_baseline = $2 WHERE id = $1 AND ramp_baseline IS NULL`, planID, raw)
+		`UPDATE purchase_plans SET ramp_baseline = $2,
+		 ramp_ledger_state = CASE WHEN ramp_ledger_protected(ramp_schedule) THEN 'trusted' ELSE ramp_ledger_state END
+		 WHERE id = $1 AND ramp_baseline IS NULL
+		 AND (COALESCE((ramp_schedule ->> 'total_steps')::int, 1) <= 1
+		      OR (ramp_ledger_state = 'new' AND COALESCE((ramp_schedule ->> 'current_step')::int, 0) = 0))`, planID, raw)
 	if err != nil {
 		return false, fmt.Errorf("failed to write ramp baseline of plan %s: %w", planID, err)
 	}
 	return tag.RowsAffected() == 1, nil
 }
 
-// GetPlanRampBaselineTx returns the frozen baseline, or nil when none was
-// written. It errors when the plan does not exist.
+// GetPlanRampBaselineTx refuses cumulative sizing when retained history is untrusted.
+// Fresh protected plans return nil until their baseline freezes under the plan lock.
 func (s *PostgresStore) GetPlanRampBaselineTx(ctx context.Context, tx pgx.Tx, planID string) (PlanRampBaseline, error) {
 	var raw []byte
-	err := tx.QueryRow(ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id = $1`, planID).Scan(&raw)
+	var state rampLedgerState
+	var total, current int
+	err := tx.QueryRow(ctx, `SELECT ramp_baseline, ramp_ledger_state,
+		COALESCE((ramp_schedule ->> 'total_steps')::int, 1),
+		COALESCE((ramp_schedule ->> 'current_step')::int, 0)
+		FROM purchase_plans WHERE id = $1`, planID).Scan(&raw, &state, &total, &current)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return nil, fmt.Errorf("%w: purchase plan %s", ErrNotFound, planID)
 	}
 	if err != nil {
 		return nil, fmt.Errorf("failed to read ramp baseline of plan %s: %w", planID, err)
 	}
+	if total > 1 && !state.allowsSizing(raw, current) {
+		return nil, ErrRampLedgerUntrusted
+	}
 	if raw == nil {
 		return nil, nil
 	}

internal/config/store_postgres_ramp_baseline_integration_test.go

@@ -159,8 +159,16 @@ func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) {
 		mutate(p)
 		require.NoError(t, f.store.UpdatePurchasePlan(ctx, p))
 	}
+	assertStale := func() {
+		var raw []byte
+		require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id=$1`, f.planID).Scan(&raw))
+		assert.Nil(t, raw)
+		readErr := f.store.WithTx(ctx, func(tx pgx.Tx) error { _, err := f.store.GetPlanRampBaselineTx(ctx, tx, f.planID); return err })
+		assert.ErrorIs(t, readErr, ErrRampLedgerUntrusted)
+	}
+	services := map[string]ServiceConfig{"ec2": {Provider: "aws", Service: "ec2", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 80}}
 	edit(func(p *PurchasePlan) {
-		p.Services = map[string]ServiceConfig{"ec2": {Provider: "aws", Service: "ec2", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 80}}
+		p.Services = services
 	})
 	freeze()
 	edit(func(p *PurchasePlan) { p.Name = "renamed" })
@@ -173,14 +181,17 @@ func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) {
 	assert.NotNil(t, f.baseline(ctx, t), "a coverage change keeps it: the target scales with coverage")
 
 	edit(func(p *PurchasePlan) { p.RampSchedule.TotalSteps = 5 })
-	assert.Nil(t, f.baseline(ctx, t), "a ramp shape change clears it")
+	assertStale()
 
+	f = newRampFixture(ctx, t, 1, 0, 4)
+	key = baselineKey(t, baselineRec(f.accounts[0], "m5.large", 100))
+	edit(func(p *PurchasePlan) { p.Services = services })
 	freeze()
 	edit(func(p *PurchasePlan) {
 		for k, svc := range p.Services {
 			svc.Term = 3
 			p.Services[k] = svc
 		}
 	})
-	assert.Nil(t, f.baseline(ctx, t), "a service identity change clears it")
+	assertStale()
 }

internal/config/store_postgres_ramp_retention_integration_test.go

@@ -0,0 +1,95 @@
+//go:build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRampRetention_NullServicesRemainEditable(t *testing.T) {
+	ctx := context.Background()
+	store := setupRampStepStore(ctx, t)
+	p := &PurchasePlan{Name: "disabled fresh", RampSchedule: RampSchedule{Type: "weekly", PercentPerStep: 25, TotalSteps: 4, StepIntervalDays: 7}}
+	require.NoError(t, store.CreatePurchasePlan(ctx, p))
+	p.Name = "renamed"
+	require.NoError(t, store.UpdatePurchasePlan(ctx, p))
+	p.RampSchedule.StepIntervalDays = 30
+	require.NoError(t, store.UpdatePurchasePlan(ctx, p))
+	var state string
+	require.NoError(t, store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, p.ID).Scan(&state))
+	require.Equal(t, "new", state)
+}
+
+func TestRampRetention_FreshEditsRequireContinuousUnheldLineage(t *testing.T) {
+	cases := []struct {
+		name, status, lineage string
+		noSubmission          bool
+		count                 int
+		want                  string
+	}{
+		{name: "fresh", want: "new"},
+		{name: "empty pending", status: "pending", want: "legacy"},
+		{name: "zero completed", status: StatusCompleted, want: "legacy"},
+		{name: "selected completed", status: StatusCompleted, count: 20, want: "legacy"},
+		{name: "unknown failed", status: StatusFailed, want: "legacy"},
+		{name: "unsubmitted", status: StatusFailed, noSubmission: true, want: "new"},
+		{name: "canceled", status: StatusCanceled, want: "new"},
+		{name: "expired", status: StatusExpired, want: "new"},
+		{name: "trusted", lineage: "trusted", want: "legacy"},
+		{name: "legacy", lineage: "legacy", want: "legacy"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			f := newRampFixture(ctx, t, 0, 0, 4)
+			if tc.status != "" {
+				var recs []RecommendationRecord
+				if tc.count > 0 {
+					recs = []RecommendationRecord{{Selected: true, Count: tc.count}}
+				}
+				f.rowWithRecs(ctx, t, 1, nil, tc.status, tc.noSubmission, recs...)
+			}
+			if tc.lineage != "" {
+				_, err := f.store.db.Exec(ctx, `UPDATE purchase_plans SET ramp_baseline='{"fixture":{"count":100}}', ramp_ledger_state=$2 WHERE id=$1`, f.planID, tc.lineage)
+				require.NoError(t, err)
+			}
+			p, err := f.store.GetPurchasePlan(ctx, f.planID)
+			require.NoError(t, err)
+			p.RampSchedule.Type, p.RampSchedule.StepIntervalDays = "custom", 30
+			require.NoError(t, f.store.UpdatePurchasePlan(ctx, p))
+			var state string
+			require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&state))
+			require.Equal(t, tc.want, state)
+		})
+	}
+}
+
+func TestRampRetention_LegacyNeverRearmsAndFreezeIsAtomic(t *testing.T) {
+	ctx := context.Background()
+	f := newRampFixture(ctx, t, 0, 0, 4)
+	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID)
+		if err != nil {
+			return err
+		}
+		wrote, err := f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{"fixture": {Count: 100}})
+		require.True(t, wrote)
+		return err
+	}))
+	_, err := f.store.db.Exec(ctx, `UPDATE purchase_plans SET ramp_baseline=NULL WHERE id=$1`, f.planID)
+	require.NoError(t, err)
+	_, err = f.store.db.Exec(ctx, `UPDATE purchase_plans SET ramp_ledger_state='new', ramp_schedule=jsonb_set(ramp_schedule,'{current_step}','0') WHERE id=$1`, f.planID)
+	require.NoError(t, err)
+	var state string
+	require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&state))
+	require.Equal(t, "legacy", state)
+	err = f.store.WithTx(ctx, func(tx pgx.Tx) error {
+		_, readErr := f.store.GetPlanRampBaselineTx(ctx, tx, f.planID)
+		return readErr
+	})
+	require.ErrorIs(t, err, ErrRampLedgerUntrusted)
+}

internal/config/store_postgres_ramp_target_met_integration_test.go

@@ -17,7 +17,7 @@ import (
 // still cannot advance through the unchanged gate.
 func TestCompleteTargetMetStepTx_RealPlan(t *testing.T) {
 	ctx := context.Background()
-	f := newRampFixture(ctx, t, 1, 1, 4)
+	f := newRampFixture(ctx, t, 1, 0, 4)
 	r := baselineRec(f.accounts[0], "m5.large", 100)
 	f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusCompleted, false, r)
 	failed := f.rowWithRecs(ctx, t, 1, &f.accounts[0], StatusFailed, false, r)
@@ -27,7 +27,10 @@ func TestCompleteTargetMetStepTx_RealPlan(t *testing.T) {
 	require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
 		_, err := f.store.LockPurchasePlanTx(ctx, tx, f.planID)
 		require.NoError(t, err)
-		_, err = f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 100}})
+		wrote, err := f.store.SetPlanRampBaselineTx(ctx, tx, f.planID, PlanRampBaseline{key: {Count: 100}})
+		require.NoError(t, err)
+		require.True(t, wrote)
+		_, err = tx.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule=jsonb_set(ramp_schedule,'{current_step}','1') WHERE id=$1`, f.planID)
 		require.NoError(t, err)
 		return f.store.CompleteTargetMetStepTx(ctx, tx, f.planID, 2, "", "target already met: target 40, committed 40")
 	}))

internal/config/store_postgres_retention_integration_test.go

@@ -190,7 +190,7 @@ func TestRampRetention_CreationLineageAndLegacyWriters(t *testing.T) {
 	f := newRampFixture(ctx, t, 0, 0, 4)
 	var state string
 	require.NoError(t, f.store.db.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&state))
-	require.Equal(t, "legacy", state)
+	require.Equal(t, "new", state)
 	legacyID := uuid.NewString()
 	_, err := f.store.db.Exec(ctx, `INSERT INTO purchase_plans(id,name,ramp_schedule) SELECT $1,'old writer',ramp_schedule FROM purchase_plans WHERE id=$2`, legacyID, f.planID)
 	require.NoError(t, err)

internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql

@@ -0,0 +1,4 @@
+-- Intentionally a no-op: the up migration cannot tell the plans it promoted
+-- from plans created as 'new', and demoting them would re-block sizing of
+-- ramps that have never bought anything.
+SELECT 1;

internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql

@@ -0,0 +1,30 @@
+-- Promote never-started legacy ramps to the retention ledger (platform#631).
+--
+-- Owner decision 2026-10-17: no backwards compatibility. A plan stamped
+-- 'legacy' by 000124's default that is a protected multi-step ramp, has not
+-- advanced (current_step = 0, no frozen baseline) and has no surviving
+-- purchase_executions row of any status is treated as having no pre-ledger
+-- history to distrust, so it becomes 'new' and sizes its first step normally.
+-- This is not proof it never bought: retention cleanup could have deleted a
+-- completed pre-000124 row of a plan whose step never advanced. That narrow
+-- risk is accepted under the owner's decision. Mid-ramp legacy plans and any
+-- plan with an execution row are deliberately left legacy.
+--
+-- guard_ramp_ledger_state forces OLD.ramp_ledger_state = 'legacy' to stay
+-- legacy on every UPDATE, so the guard is disabled for this one statement.
+-- DISABLE TRIGGER names only that trigger on this table (unlike
+-- session_replication_role, it needs no superuser and leaves
+-- retain_ramp_ledger_execution and every other trigger live). The migration
+-- runs in one transaction, so the guard is re-enabled before commit and a
+-- failure rolls both back.
+ALTER TABLE purchase_plans DISABLE TRIGGER guard_ramp_ledger_state;
+
+UPDATE purchase_plans p
+SET ramp_ledger_state = 'new'
+WHERE p.ramp_ledger_state = 'legacy'
+  AND ramp_ledger_protected(p.ramp_schedule)
+  AND COALESCE((p.ramp_schedule ->> 'current_step')::INT, 0) = 0
+  AND p.ramp_baseline IS NULL
+  AND NOT EXISTS (SELECT 1 FROM purchase_executions e WHERE e.plan_id = p.id);
+
+ALTER TABLE purchase_plans ENABLE TRIGGER guard_ramp_ledger_state;

internal/purchase/notifications.go

@@ -292,21 +292,29 @@ func adoptableRootRow(rows []config.PurchaseExecution) (*config.PurchaseExecutio
 }
 
 // failExistingPlanStep fails a pending/notified plan-step row that will never
-// carry recommendations, with reason. The status change is a compare-and-set,
+// be bought, with reason. The CAS pins the recommendations to what the caller
+// saw (empty for a bare row, non-empty for a refused attached one). The status change is a compare-and-set,
 // so a human approving the row between this tick's read and now wins: the row
 // is no longer awaiting approval and the tick skips it, instead of an upsert
 // from the stale read overwriting the approval. A won CAS returns the fresh
 // row, which is the one the error text is written onto.
 func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.PurchaseExecution, reason string) error {
-	failed, err := m.config.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, true)
+	failed, err := m.config.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"pending", "notified"}, "failed", nil, len(exec.Recommendations) == 0)
 	if errors.Is(err, config.ErrNotFound) || errors.Is(err, config.ErrExecutionNotInExpectedStatus) {
 		return fmt.Errorf("%w: %s changed state before it could be failed", errExecutionNotNotifiable, exec.ExecutionID)
 	}
 	if err != nil {
 		return fmt.Errorf("failed to fail plan step %d of plan %s: %w", exec.StepNumber, exec.PlanID, err)
 	}
 	stampFailedPlanStep(failed, reason)
-	if err := m.config.SavePurchaseExecution(ctx, failed); err != nil {
+	// The row is failed with NoSubmission, so its suppressions are released as
+	// for any proven not-submitted failure (saveTerminalExecution); a bare row has none.
+	if err := m.config.WithTx(ctx, func(tx pgx.Tx) error {
+		if err := m.config.SavePurchaseExecutionTx(ctx, tx, failed); err != nil {
+			return err
+		}
+		return m.config.DeleteSuppressionsByExecutionTx(ctx, tx, failed.ExecutionID)
+	}); err != nil {
 		return fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", failed.StepNumber, failed.PlanID, err)
 	}
 	return bareStepFailure(failed.PlanID, failed.StepNumber, reason)

internal/purchase/plan_step_allocation_integration_test.go

@@ -20,15 +20,27 @@ func testEqualCountAttach(t *testing.T, count int, want []int) {
 	}
 	_, err := f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_schedule = ramp_schedule || jsonb_build_object('allocation_mode','equal_count','total_steps',$2::int,'percent_per_step',$3::float8), services = jsonb_set(services,'{ec2,coverage}','100') WHERE id=$1`, f.planID, count, percent)
 	require.NoError(t, err)
+	var ledger string
+	require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, f.planID).Scan(&ledger))
+	require.Equal(t, "new", ledger)
 	f.shownSuppressed(97)
 	for i, n := range want {
 		exec, tickErr := f.tick()
 		require.NoError(t, tickErr, "exact equal-count attachment")
 		assert.Equal(t, n, totalCount(exec), "exact equal-count allocation step %d", i+1)
 		f.buy(i + 1)
 	}
-	b, err := f.baseline()
-	require.NoError(t, err)
+	var b config.PlanRampBaseline
+	if f.plan().RampSchedule.IsComplete() {
+		require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id=$1`, f.planID).Scan(&b))
+		require.ErrorIs(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error {
+			_, readErr := f.store.GetPlanRampBaselineTx(f.ctx, tx, f.planID)
+			return readErr
+		}), config.ErrRampLedgerUntrusted)
+	} else {
+		b, err = f.baseline()
+		require.NoError(t, err)
+	}
 	for _, q := range b {
 		assert.Equal(t, 97, q.Count)
 	}

internal/purchase/plan_step_attach.go

@@ -96,6 +96,12 @@ func (m *Manager) claimFromRows(ctx context.Context, tx pgx.Tx, locked *config.P
 			return claim, nil
 		}
 		if len(root.Recommendations) > 0 {
+			if _, err := m.config.GetPlanRampBaselineTx(ctx, tx, locked.ID); err != nil {
+				if errors.Is(err, config.ErrRampLedgerUntrusted) {
+					return m.failStep(ctx, tx, locked, claim, root, err.Error())
+				}
+				return stepClaim{}, err
+			}
 			if locked.NextExecutionDate.Before(time.Now()) {
 				return stepClaim{}, fmt.Errorf("%w: step %d of plan %s already has a row and the plan date has passed", errExecutionNotNotifiable, claim.number, locked.ID)
 			}
@@ -159,16 +165,14 @@ func (m *Manager) attachStep(ctx context.Context, tx pgx.Tx, locked *config.Purc
 		ids = append(ids, accounts[i].ID)
 	}
 	res, err := m.resolveCumulative(ctx, tx, locked, claim.number, ids)
-	if errors.Is(err, planstep.ErrNoMatch) {
+	switch {
+	case errors.Is(err, planstep.ErrNoMatch):
 		return m.noMatch(ctx, tx, locked, claim, root, res.Skipped)
-	}
-	if errors.Is(err, planstep.ErrTargetMet) {
+	case errors.Is(err, planstep.ErrTargetMet):
 		return m.targetMet(ctx, tx, locked, claim, root, res)
-	}
-	if errors.Is(err, planstep.ErrUnsupportedPlan) {
+	case errors.Is(err, planstep.ErrUnsupportedPlan), errors.Is(err, config.ErrRampLedgerUntrusted):
 		return m.failStep(ctx, tx, locked, claim, root, err.Error())
-	}
-	if err != nil {
+	case err != nil:
 		return stepClaim{}, fmt.Errorf("plan %s step %d: %w", locked.ID, claim.number, err)
 	}
 	if len(res.Skipped) > 0 {
@@ -209,9 +213,13 @@ func (m *Manager) resolveCumulative(ctx context.Context, tx pgx.Tx, locked *conf
 	if step != 1 || baseline != nil || len(res.Baseline) == 0 {
 		return res, err
 	}
-	if _, setErr := m.config.SetPlanRampBaselineTx(ctx, tx, locked.ID, res.Baseline); setErr != nil {
+	wrote, setErr := m.config.SetPlanRampBaselineTx(ctx, tx, locked.ID, res.Baseline)
+	if setErr != nil {
 		return planstep.Result{}, setErr
 	}
+	if !wrote {
+		return planstep.Result{}, config.ErrRampLedgerUntrusted
+	}
 	if errors.Is(err, planstep.ErrNoMatch) {
 		return m.resolver.Resolve(ctx, locked, ids, step, res.Baseline, committed)
 	}