diff --git a/internal/config/store_postgres_plan_writes.go b/internal/config/store_postgres_plan_writes.go
index 6fb99f1c..e4d1a96f 100644
--- a/internal/config/store_postgres_plan_writes.go
+++ b/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
 	`
 
diff --git a/internal/config/store_postgres_ramp_baseline.go b/internal/config/store_postgres_ramp_baseline.go
index 6611773b..6c87cbcd 100644
--- a/internal/config/store_postgres_ramp_baseline.go
+++ b/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
 	}
diff --git a/internal/config/store_postgres_ramp_baseline_integration_test.go b/internal/config/store_postgres_ramp_baseline_integration_test.go
index 9421c69c..f37fcd1f 100644
--- a/internal/config/store_postgres_ramp_baseline_integration_test.go
+++ b/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,8 +181,11 @@ 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 {
@@ -182,5 +193,5 @@ func TestPlanRampBaseline_ClearedOnlyByServiceOrRampShapeEdit(t *testing.T) {
 			p.Services[k] = svc
 		}
 	})
-	assert.Nil(t, f.baseline(ctx, t), "a service identity change clears it")
+	assertStale()
 }
diff --git a/internal/config/store_postgres_ramp_retention_integration_test.go b/internal/config/store_postgres_ramp_retention_integration_test.go
new file mode 100644
index 00000000..3c1f2cda
--- /dev/null
+++ b/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)
+}
diff --git a/internal/config/store_postgres_ramp_target_met_integration_test.go b/internal/config/store_postgres_ramp_target_met_integration_test.go
index d062ec89..fe6c8ced 100644
--- a/internal/config/store_postgres_ramp_target_met_integration_test.go
+++ b/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")
 	}))
diff --git a/internal/config/store_postgres_retention_integration_test.go b/internal/config/store_postgres_retention_integration_test.go
index ad964db4..49b698a5 100644
--- a/internal/config/store_postgres_retention_integration_test.go
+++ b/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)
diff --git a/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql b/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.down.sql
new file mode 100644
index 00000000..78c35476
--- /dev/null
+++ b/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;
diff --git a/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql b/internal/database/postgres/migrations/000127_never_started_legacy_ramps_to_new.up.sql
new file mode 100644
index 00000000..26506513
--- /dev/null
+++ b/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;
diff --git a/internal/purchase/notifications.go b/internal/purchase/notifications.go
index 0cf946ba..5b34502f 100644
--- a/internal/purchase/notifications.go
+++ b/internal/purchase/notifications.go
@@ -292,13 +292,14 @@ 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)
 	}
@@ -306,7 +307,14 @@ func (m *Manager) failExistingPlanStep(ctx context.Context, exec *config.Purchas
 		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)
diff --git a/internal/purchase/plan_step_allocation_integration_test.go b/internal/purchase/plan_step_allocation_integration_test.go
index 9e218f2b..aaaed46c 100644
--- a/internal/purchase/plan_step_allocation_integration_test.go
+++ b/internal/purchase/plan_step_allocation_integration_test.go
@@ -20,6 +20,9 @@ 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()
@@ -27,8 +30,17 @@ func testEqualCountAttach(t *testing.T, count int, want []int) {
 		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)
 	}
diff --git a/internal/purchase/plan_step_attach.go b/internal/purchase/plan_step_attach.go
index 1008b9e8..31ee87b7 100644
--- a/internal/purchase/plan_step_attach.go
+++ b/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)
 	}
diff --git a/internal/purchase/plan_step_cumulative_integration_test.go b/internal/purchase/plan_step_cumulative_integration_test.go
index 7faa80a1..77ec5219 100644
--- a/internal/purchase/plan_step_cumulative_integration_test.go
+++ b/internal/purchase/plan_step_cumulative_integration_test.go
@@ -4,6 +4,7 @@ package purchase
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"testing"
@@ -64,9 +65,15 @@ func cumRec(account, typ string, count int) config.RecommendationRecord {
 const cumCoverage = 80
 
 func newCumFixture(t *testing.T, stored ...int) *cumFixture {
+	t.Helper()
+	store, pool := rampStepStore(context.Background(), t)
+	return newCumFixtureOn(t, store, pool, stored...)
+}
+
+// newCumFixtureOn builds the fixture on an already migrated store.
+func newCumFixtureOn(t *testing.T, store *config.PostgresStore, pool *pgxpool.Pool, stored ...int) *cumFixture {
 	t.Helper()
 	ctx := context.Background()
-	store, pool := rampStepStore(ctx, t)
 	attachGlobal(ctx, t, store)
 	acct := &config.CloudAccount{Name: "cum-acct", Provider: "aws", ExternalID: "111111111110", Enabled: true,
 		AWSAuthMode: "role_arn", AWSRoleARN: "arn:aws:iam::111111111110:role/cudly"}
@@ -170,8 +177,10 @@ func TestCumulativeAttach_WeeklyRampBuysTwentyEachStep(t *testing.T) {
 				f.buy(step)
 			}
 			assert.Equal(t, 4, f.plan().RampSchedule.CurrentStep)
-			base, err := f.baseline()
-			require.NoError(t, err)
+			var raw []byte
+			require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT ramp_baseline FROM purchase_plans WHERE id=$1`, f.planID).Scan(&raw))
+			var base config.PlanRampBaseline
+			require.NoError(t, json.Unmarshal(raw, &base))
 			require.Len(t, base, 1)
 			for _, q := range base {
 				assert.Equal(t, 100, q.Count, "frozen at step 1 from shown")
diff --git a/internal/purchase/plan_step_ledger_migration_integration_test.go b/internal/purchase/plan_step_ledger_migration_integration_test.go
new file mode 100644
index 00000000..714971fb
--- /dev/null
+++ b/internal/purchase/plan_step_ledger_migration_integration_test.go
@@ -0,0 +1,109 @@
+//go:build integration
+
+package purchase
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+const neverStartedMigration = "000127_never_started_legacy_ramps_to_new"
+
+// stagedMigrations returns a directory holding every migration but the one
+// under test, and a func that adds it, so the test runs the real runner twice.
+func stagedMigrations(t *testing.T) (dir string, addTarget func()) {
+	t.Helper()
+	src := rampMigrationsPath()
+	dir = t.TempDir()
+	entries, err := os.ReadDir(src)
+	require.NoError(t, err)
+	copyFile := func(name string) {
+		b, readErr := os.ReadFile(filepath.Join(src, name))
+		require.NoError(t, readErr)
+		require.NoError(t, os.WriteFile(filepath.Join(dir, name), b, 0o600))
+	}
+	for _, e := range entries {
+		if filepath.Ext(e.Name()) == ".sql" && !strings.HasPrefix(e.Name(), neverStartedMigration) {
+			copyFile(e.Name())
+		}
+	}
+	return dir, func() {
+		copyFile(neverStartedMigration + ".up.sql")
+		copyFile(neverStartedMigration + ".down.sql")
+	}
+}
+
+func TestMigration_NeverStartedLegacyRampsBecomeNew(t *testing.T) {
+	ctx := context.Background()
+	container, err := testhelpers.SetupPostgresContainer(ctx, t)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = container.Cleanup(ctx) })
+	dir, addTarget := stagedMigrations(t)
+	require.NoError(t, migrations.RunMigrations(ctx, container.DB.Pool(), dir, "", ""))
+	store, pool := config.NewPostgresStore(container.DB), container.DB.Pool()
+
+	f := newCumFixtureOn(t, store, pool, 100) // plan (a): never started, no executions
+	clone := func(name string, total, current int) string {
+		p := f.plan()
+		p.ID, p.Name = "", name
+		p.RampSchedule.TotalSteps = total
+		require.NoError(t, store.CreatePurchasePlan(ctx, p))
+		_, execErr := pool.Exec(ctx, `UPDATE purchase_plans SET ramp_schedule=jsonb_set(ramp_schedule,'{current_step}',to_jsonb($2::int)) WHERE id=$1`, p.ID, current)
+		require.NoError(t, execErr)
+		return p.ID
+	}
+	withExec := clone("b-has-execution", 4, 0)
+	require.NoError(t, store.SavePurchaseExecution(ctx, &config.PurchaseExecution{PlanID: withExec, ExecutionID: uuid.NewString(), Status: config.StatusFailed, NoSubmission: true, StepNumber: 1, ScheduledDate: time.Now()}))
+	midRamp := clone("c-mid-ramp", 4, 2)
+	singleStep := clone("d-single-step", 1, 0)
+	trusted := clone("e-trusted", 4, 0)
+	withBaseline := clone("f-legacy-baseline", 4, 0)
+
+	// Force the states the old writer left behind; the guard would otherwise rewrite them.
+	_, err = pool.Exec(ctx, `ALTER TABLE purchase_plans DISABLE TRIGGER guard_ramp_ledger_state`)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, `UPDATE purchase_plans SET ramp_ledger_state='legacy' WHERE id <> $1`, trusted)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, `UPDATE purchase_plans SET ramp_baseline='{"k":{"count":1}}'::jsonb WHERE id = $1`, withBaseline)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, `UPDATE purchase_plans SET ramp_ledger_state='trusted', ramp_baseline='{"k":{"count":1}}'::jsonb WHERE id = $1`, trusted)
+	require.NoError(t, err)
+	_, err = pool.Exec(ctx, `ALTER TABLE purchase_plans ENABLE TRIGGER guard_ramp_ledger_state`)
+	require.NoError(t, err)
+
+	addTarget()
+	require.NoError(t, migrations.RunMigrations(ctx, container.DB.Pool(), dir, "", ""))
+
+	state := func(id string) (s string) {
+		require.NoError(t, pool.QueryRow(ctx, `SELECT ramp_ledger_state FROM purchase_plans WHERE id=$1`, id).Scan(&s))
+		return s
+	}
+	require.Equal(t, "new", state(f.planID), "never started, no executions")
+	require.Equal(t, "legacy", state(withExec), "an execution row of any status keeps it legacy")
+	require.Equal(t, "legacy", state(midRamp), "mid-ramp plans are not touched")
+	require.Equal(t, "legacy", state(singleStep), "single-step plans are not protected")
+	require.Equal(t, "legacy", state(withBaseline), "a frozen baseline is never discarded")
+	require.Equal(t, "trusted", state(trusted))
+	var enabled string
+	require.NoError(t, pool.QueryRow(ctx, `SELECT tgenabled::text FROM pg_trigger WHERE tgname='guard_ramp_ledger_state'`).Scan(&enabled))
+	require.Equal(t, "O", enabled, "the guard is live again")
+
+	rec := cumRec(f.account, "m5.t0", 100)
+	f.src.shown = func() []config.RecommendationRecord { return []config.RecommendationRecord{rec} }
+	exec, err := f.tick()
+	require.NoError(t, err, "the promoted ramp sizes step 1 normally")
+	require.NotEmpty(t, exec.Recommendations)
+	base, err := f.baseline()
+	require.NoError(t, err)
+	require.NotEmpty(t, base, "step 1 froze a baseline")
+}
diff --git a/internal/purchase/plan_step_retention_integration_test.go b/internal/purchase/plan_step_retention_integration_test.go
new file mode 100644
index 00000000..f1ea65cd
--- /dev/null
+++ b/internal/purchase/plan_step_retention_integration_test.go
@@ -0,0 +1,236 @@
+//go:build integration
+
+package purchase
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCumulativeAttach_RetentionKeepsCommittedCapacity(t *testing.T) {
+	f := newCumFixture(t, 100)
+	p := f.plan()
+	p.RampSchedule.Type, p.RampSchedule.StepIntervalDays = "custom", 30
+	require.NoError(t, f.store.UpdatePurchasePlan(f.ctx, p))
+	f.src.shown = func() []config.RecommendationRecord {
+		return []config.RecommendationRecord{cumRec(f.account, "m5.t0", 100)}
+	}
+	first, err := f.tick()
+	require.NoError(t, err)
+	require.Equal(t, 20, totalCount(first))
+	f.buy(1)
+	f.src.shown = func() []config.RecommendationRecord {
+		return []config.RecommendationRecord{cumRec(f.account, "m5.t0", 80)}
+	}
+	before, err := f.baseline()
+	require.NoError(t, err)
+	require.Len(t, before, 1)
+	baselineCount := 0
+	for _, q := range before {
+		baselineCount = q.Count
+		require.Equal(t, 100, q.Count)
+	}
+	require.NoError(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error {
+		_, updateErr := tx.Exec(f.ctx, `UPDATE purchase_executions SET scheduled_date=NOW()-INTERVAL '31 days' WHERE plan_id=$1 AND step_number=1`, f.planID)
+		return updateErr
+	}))
+	deleted, err := f.store.CleanupOldExecutions(f.ctx, 30)
+	require.NoError(t, err)
+	require.Zero(t, deleted)
+	after, err := f.baseline()
+	require.NoError(t, err)
+	require.Equal(t, before, after)
+	var committed map[string]config.PlanRampQuantity
+	require.NoError(t, f.store.WithTx(f.ctx, func(tx pgx.Tx) error {
+		_, lockErr := f.store.LockPurchasePlanTx(f.ctx, tx, f.planID)
+		if lockErr != nil {
+			return lockErr
+		}
+		committed, err = f.store.ListPlanCommittedTx(f.ctx, tx, f.planID)
+		return err
+	}))
+	heldCount := 0
+	for _, q := range committed {
+		heldCount = q.Count
+		require.Equal(t, 20, q.Count)
+	}
+	require.Len(t, committed, 1)
+	next, err := f.tick()
+	require.NoError(t, err)
+	require.Equal(t, 20, totalCount(next), "target 40 less the retained first 20")
+	t.Logf("synthetic quantities: baseline=%d first=%d retained=%d next_target=40 next_attached=%d", baselineCount, totalCount(first), heldCount, totalCount(next))
+}
+
+func TestCumulativeAttach_FreshEditOverlapsInvisibleHeldInsertion(t *testing.T) {
+	f := newCumFixture(t, 100)
+	tx, err := f.pool.Begin(f.ctx)
+	require.NoError(t, err)
+	defer tx.Rollback(f.ctx)
+	held := cumRec(f.account, "m5.t0", 20)
+	held.Selected = true
+	require.NoError(t, f.store.SavePurchaseExecutionTx(f.ctx, tx, &config.PurchaseExecution{
+		PlanID: f.planID, ExecutionID: uuid.NewString(), Status: "pending", StepNumber: 2,
+		ScheduledDate: time.Now().Add(-31 * 24 * time.Hour), TTL: time.Now().Add(-31 * 24 * time.Hour).Unix(),
+		Recommendations: []config.RecommendationRecord{held},
+	}))
+	bounded, cancel := context.WithTimeout(f.ctx, 2*time.Second)
+	defer cancel()
+	_, err = f.pool.Exec(bounded, `UPDATE purchase_plans SET ramp_schedule=jsonb_set(jsonb_set(ramp_schedule,'{type}','"custom"'),'{step_interval_days}','30') WHERE id=$1`, f.planID)
+	require.NoError(t, err, "old writer edits while the new held row is still invisible")
+	require.NoError(t, tx.Commit(f.ctx))
+	deleted, err := f.store.CleanupOldExecutions(f.ctx, 30)
+	require.NoError(t, err)
+	require.Zero(t, deleted)
+	f.src.shown = func() []config.RecommendationRecord {
+		return []config.RecommendationRecord{cumRec(f.account, "m5.t0", 100)}
+	}
+	_, err = f.tick()
+	require.ErrorIs(t, err, errExecutionNotNotifiable)
+	rows := f.stepRows(1)
+	require.Len(t, rows, 1)
+	require.Empty(t, rows[0].Recommendations, "first target20 is already held, so the freeze must not attach another20")
+	require.NotNil(t, rows[0].PreApprovalSkipReason)
+	require.Equal(t, config.PreApprovalSkipRampTargetMet, *rows[0].PreApprovalSkipReason)
+	base, err := f.baseline()
+	require.NoError(t, err)
+	require.Len(t, base, 1)
+}
+
+func TestCumulativeAttach_UntrustedHistoryCannotSizeOrNotify(t *testing.T) {
+	cases := []struct {
+		name, rootStatus string
+		progress, held   int
+		clear            bool
+	}{
+		{name: "all prior history missing", progress: 1},
+		{name: "partial prior history missing", progress: 1, held: 10},
+		{name: "progressed baseline missing", progress: 1, clear: true},
+		{name: "legacy step zero baseline missing", clear: true},
+		{name: "nonempty pending root", progress: 1, rootStatus: "pending"},
+		{name: "nonempty notified root", progress: 1, rootStatus: "notified"},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newCumFixture(t, 100)
+			rec := cumRec(f.account, "m5.t0", 100)
+			key, err := config.RecIdentityKey(&rec)
+			require.NoError(t, err)
+			raw, err := json.Marshal(config.PlanRampBaseline{key: {Count: 100}})
+			require.NoError(t, err)
+			_, err = f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_baseline=$2, ramp_schedule=jsonb_set(ramp_schedule,'{current_step}',to_jsonb($3::int)) WHERE id=$1`, f.planID, raw, tc.progress)
+			require.NoError(t, err, "old writer freezes without a trust transition")
+			if tc.clear {
+				_, err = f.pool.Exec(f.ctx, `UPDATE purchase_plans SET ramp_baseline=NULL WHERE id=$1`, f.planID)
+				require.NoError(t, err)
+			}
+			if tc.held > 0 {
+				rec.Count, rec.Selected = tc.held, true
+				require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: config.StatusCompleted, StepNumber: 1, ScheduledDate: time.Now(), Recommendations: []config.RecommendationRecord{rec}}))
+			}
+			var rootID string
+			if tc.rootStatus != "" {
+				rec.Selected = true
+				rootID = uuid.NewString()
+				require.NoError(t, f.store.SavePurchaseExecution(f.ctx, &config.PurchaseExecution{PlanID: f.planID, ExecutionID: rootID, Status: tc.rootStatus, StepNumber: tc.progress + 1, ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}}))
+				for _, id := range []string{rootID} {
+					require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: id, AccountID: "sup-acct", Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)}))
+				}
+				require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: uuid.NewString(), AccountID: "other-acct", Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)}))
+			}
+			calls := 0
+			f.src.shown = func() []config.RecommendationRecord { calls++; return []config.RecommendationRecord{rec} }
+			_, err = f.tick()
+			require.Error(t, err)
+			rows := f.stepRows(tc.progress + 1)
+			require.Len(t, rows, 1)
+			if tc.rootStatus == "" {
+				require.Equal(t, config.StatusFailed, rows[0].Status)
+				require.Contains(t, rows[0].Error, "cannot be verified")
+				require.Empty(t, rows[0].Recommendations)
+			} else {
+				require.ErrorIs(t, err, errExecutionNotNotifiable, "a durable refusal is a quiet skip, not a raw error")
+				require.NotErrorIs(t, err, config.ErrRampLedgerUntrusted)
+				require.Equal(t, config.StatusFailed, rows[0].Status)
+				require.True(t, rows[0].NoSubmission)
+				require.Contains(t, rows[0].Error, config.ErrRampLedgerUntrusted.Error())
+				require.NotEmpty(t, rows[0].Recommendations, "the attached recommendations stay on the failed row")
+				var released, others int
+				require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM purchase_suppressions WHERE execution_id = $1`, rootID).Scan(&released))
+				require.Zero(t, released, "a proven not-submitted failure releases its suppressions")
+				require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM purchase_suppressions WHERE account_id = 'other-acct'`).Scan(&others))
+				require.Equal(t, 1, others, "another execution's suppressions are untouched")
+			}
+			require.False(t, f.m.sendPlanNotification(f.ctx, f.plan()))
+			require.Zero(t, calls, "no new sizing or resizing")
+			require.Zero(t, f.m.email.(*countingSender).sends)
+			require.Zero(t, suppressedTotal(f.ctx, t, f.pool))
+			require.Empty(t, rows[0].ApprovalToken)
+		})
+	}
+}
+
+func TestCumulativeAttach_AuthorizedLegacyQuantityRemainsImmutable(t *testing.T) {
+	for _, status := range []string{"scheduled", "approved"} {
+		t.Run(status, func(t *testing.T) {
+			ctx := context.Background()
+			f := newRampStepFixture(ctx, t)
+			require.NoError(t, f.store.SetPlanAccounts(ctx, f.plan.ID, []string{f.accounts[0].ID}))
+			root := f.saveRootExecution(ctx, t, 3)
+			root.Status = status
+			root.CloudAccountID = &f.accounts[0].ID
+			root.Recommendations = root.Recommendations[:1]
+			root.Recommendations[0].Count = 7
+			due := time.Now().Add(-time.Minute)
+			root.ScheduledExecutionAt = &due
+			require.NoError(t, f.store.SavePurchaseExecution(ctx, root))
+			if status == "scheduled" {
+				result, fireErr := f.manager.FireScheduledDelayedPurchases(ctx)
+				require.NoError(t, fireErr)
+				require.Equal(t, 1, result.Fired)
+				require.Zero(t, result.Errored)
+			} else {
+				require.NoError(t, f.execute(ctx, root))
+			}
+			f.mu.Lock()
+			counts := append([]int(nil), f.counts...)
+			f.mu.Unlock()
+			require.Equal(t, []int{7}, counts, "the recording provider receives the authorized frozen quantity")
+			resolver := &fakeResolver{recs: root.Recommendations}
+			f.manager.resolver = resolver
+			err := f.store.WithTx(ctx, func(tx pgx.Tx) error {
+				p, lockErr := f.store.LockPurchasePlanTx(ctx, tx, f.plan.ID)
+				if lockErr != nil {
+					return lockErr
+				}
+				_, resolveErr := f.manager.resolveCumulative(ctx, tx, p, p.RampSchedule.CurrentStep+1, []string{f.accounts[0].ID})
+				return resolveErr
+			})
+			require.ErrorIs(t, err, config.ErrRampLedgerUntrusted, "the next step may not size from unverifiable history")
+			require.Zero(t, resolver.calls.Load())
+		})
+	}
+}
+
+func TestFailExistingPlanStep_LostRaceKeepsSuppressions(t *testing.T) {
+	f := newCumFixture(t, 100)
+	rec := cumRec(f.account, "m5.t0", 100)
+	rec.Selected = true
+	root := &config.PurchaseExecution{PlanID: f.planID, ExecutionID: uuid.NewString(), Status: "pending", StepNumber: 1, ScheduledDate: time.Now().Add(40 * 24 * time.Hour), Recommendations: []config.RecommendationRecord{rec}}
+	require.NoError(t, f.store.SavePurchaseExecution(f.ctx, root))
+	require.NoError(t, f.store.CreateSuppression(f.ctx, &config.PurchaseSuppression{ExecutionID: root.ExecutionID, AccountID: "sup-acct", Provider: "aws", Service: "ec2", Region: "us-east-1", ResourceType: "m5.t0", SuppressedCount: 1, ExpiresAt: time.Now().Add(time.Hour)}))
+	_, err := f.pool.Exec(f.ctx, `UPDATE purchase_executions SET status='approved' WHERE execution_id=$1`, root.ExecutionID)
+	require.NoError(t, err, "a human approves after the tick read the row")
+
+	err = f.m.failExistingPlanStep(f.ctx, root, "reason")
+	require.ErrorIs(t, err, errExecutionNotNotifiable)
+	var n int
+	require.NoError(t, f.pool.QueryRow(f.ctx, `SELECT COUNT(*) FROM purchase_suppressions WHERE execution_id = $1`, root.ExecutionID).Scan(&n))
+	require.Equal(t, 1, n, "the approved row keeps its suppressions")
+}
diff --git a/internal/purchase/ramp_step_progress_integration_test.go b/internal/purchase/ramp_step_progress_integration_test.go
index c8889e9c..8553c642 100644
--- a/internal/purchase/ramp_step_progress_integration_test.go
+++ b/internal/purchase/ramp_step_progress_integration_test.go
@@ -68,6 +68,7 @@ type rampStepFixture struct {
 
 	mu        sync.Mutex
 	purchased []string // one entry per commitment that reached the fake cloud
+	counts    []int
 
 	// onPurchase, when set, runs inside each fake provider call (issue #829 tests).
 	onPurchase func()
@@ -156,6 +157,7 @@ func newRampStepFixture(ctx context.Context, t *testing.T) *rampStepFixture {
 			opts := args.Get(2).(common.PurchaseOptions)
 			f.mu.Lock()
 			f.purchased = append(f.purchased, opts.IdempotencyToken)
+			f.counts = append(f.counts, args.Get(1).(common.Recommendation).Count)
 			f.mu.Unlock()
 			if f.onPurchase != nil {
 				f.onPurchase()
diff --git a/known_issues/36_plan_step_fail_cas_not_atomic.md b/known_issues/36_plan_step_fail_cas_not_atomic.md
new file mode 100644
index 00000000..9a654d04
--- /dev/null
+++ b/known_issues/36_plan_step_fail_cas_not_atomic.md
@@ -0,0 +1,30 @@
+# Known Issues: plan step failure is not atomic with its stamp and suppression release
+
+> **Audit status (2026-10-17):** `1 needs triage · 0 resolved`
+
+## LOW: crash between the failed CAS and the stamp tx strands the row
+
+**Files:** `internal/purchase/notifications.go` (`failExistingPlanStep`),
+`internal/config/store_postgres.go` (`TransitionExecutionStatusPinned`),
+`internal/purchase/plan_step_attach.go` (`claimFromRows`, `failStep`)
+
+**Description:** `failExistingPlanStep` first flips the row to `failed` with a
+pinned compare-and-set, then saves the `Error` and `NoSubmission` stamp and
+deletes the row's suppressions in a second transaction. The CAS cannot join a
+caller transaction, so a crash or DB error between the two leaves the row
+`failed` with no `Error`, no `NoSubmission`, and its suppressions still in
+place until their grace period expires. The `Error`/`NoSubmission` half of the
+gap predates the suppression release; the release makes it slightly wider.
+
+**Why it fails closed:** `adoptableRootRow` only adopts pending/notified roots,
+so a failed row is skipped on every later tick and the step never gets a second
+root. Nothing is bought twice. The cost is a step that stays failed without a
+recorded reason and recommendations hidden until the suppressions expire.
+
+**Fix direction:** add a Tx variant of `TransitionExecutionStatusPinned` so the
+CAS, the stamp and `DeleteSuppressionsByExecutionTx` commit together.
+
+**Why deferred:** out of scope for the retention-ledger runtime change (#905);
+the window is a crash between two statements and the failure mode is safe.
+
+**Status:** Open. Tracked in #1052.