diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go
index 5ff192066..7fd34783a 100644
--- a/internal/api/handler_purchases.go
+++ b/internal/api/handler_purchases.go
@@ -2556,6 +2556,9 @@ func (h *Handler) loadAndValidateRetryRequest(ctx context.Context, req *events.L
 	if err := checkRetryEligibilityGates(failedExec, req); err != nil {
 		return nil, nil, err
 	}
+	if err := h.checkRedshiftRetryGate(ctx, failedExec); err != nil {
+		return nil, nil, err
+	}
 
 	return failedExec, session, nil
 }
@@ -2784,6 +2787,7 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config.
 		CapacityPercent:        failedExec.CapacityPercent,
 		CreatedByUserID:        resolveCreatorUserID(session),
 		RetryAttemptN:          failedExec.RetryAttemptN + 1,
+		FirstDispatchAt:        retrySuccessorFirstDispatchAt(failedExec),
 	}
 
 	graceCfg, err := h.gracePeriodConfig(ctx)
diff --git a/internal/api/handler_purchases_redshift_retry.go b/internal/api/handler_purchases_redshift_retry.go
new file mode 100644
index 000000000..520e4814b
--- /dev/null
+++ b/internal/api/handler_purchases_redshift_retry.go
@@ -0,0 +1,45 @@
+package api
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/purchase"
+)
+
+// checkRedshiftRetryGate refuses a Retry that could still double-buy: permanently
+// (unknown or past-horizon dispatch) or until AWS settles. Force never applies.
+func (h *Handler) checkRedshiftRetryGate(ctx context.Context, failedExec *config.PurchaseExecution) error {
+	if !purchase.IsRedshiftExecution(failedExec) {
+		return nil
+	}
+	clock, err := h.config.GetRedshiftRetryClock(ctx, failedExec.ExecutionID)
+	if err != nil {
+		return fmt.Errorf("read redshift retry clock: %w", err)
+	}
+	if reason := purchase.RedshiftRetryPermanentRefusal(failedExec, clock.SinceStamp); reason != "" {
+		return NewClientErrorWithDetails(409,
+			"this purchase cannot be retried safely: "+reason,
+			map[string]any{"ops_hint": reason, "redrive_unsafe": true})
+	}
+	if wait := purchase.RedshiftRetryWait(failedExec, clock.SinceUpdate); wait > 0 {
+		secs := int((wait + time.Second - 1) / time.Second)
+		return NewClientErrorWithHeaders(409,
+			"this Redshift purchase failed too recently to retry safely; retry after the wait",
+			map[string]any{"retry_after_seconds": secs},
+			map[string]string{"Retry-After": strconv.Itoa(secs)})
+	}
+	return nil
+}
+
+// retrySuccessorFirstDispatchAt carries the lineage stamp to a retry successor;
+// a never-submitted attempt 0 has none and the successor mints its own.
+func retrySuccessorFirstDispatchAt(failedExec *config.PurchaseExecution) *time.Time {
+	if purchase.NeverSubmitted(failedExec) {
+		return nil
+	}
+	return clonePtr(failedExec.FirstDispatchAt)
+}
diff --git a/internal/api/handler_purchases_redshift_retry_test.go b/internal/api/handler_purchases_redshift_retry_test.go
new file mode 100644
index 000000000..7083f17e3
--- /dev/null
+++ b/internal/api/handler_purchases_redshift_retry_test.go
@@ -0,0 +1,158 @@
+package api
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+const redshiftRetryExecID = "22222222-3333-4444-5555-666666666701"
+
+func redshiftFailedRow(attempt int, noSubmission bool) *config.PurchaseExecution {
+	failed := redriveFailedRow(redshiftRetryExecID, "aws", "redshift")
+	failed.RetryAttemptN = attempt
+	failed.NoSubmission = noSubmission
+	return failed
+}
+
+func gateWithClock(t *testing.T, failed *config.PurchaseExecution, clock config.RedshiftRetryClock) *Handler {
+	t.Helper()
+	store := new(MockConfigStore)
+	store.On("GetRedshiftRetryClock", mock.Anything, failed.ExecutionID).Return(clock, nil)
+	return &Handler{config: store}
+}
+
+func ageOf(d time.Duration) *time.Duration { return &d }
+
+func TestCheckRedshiftRetryGate_LegacyStamplessPermanent(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: time.Hour})
+
+	err := h.checkRedshiftRetryGate(context.Background(), failed)
+
+	assertRedriveRefused(t, err, "no first dispatch time recorded")
+	ce, _ := IsClientError(err)
+	assert.NotContains(t, ce.Details(), "retry_after_seconds")
+}
+
+func TestCheckRedshiftRetryGate_PastHorizonPermanent(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: time.Hour, SinceStamp: ageOf(24 * time.Hour)})
+
+	assertRedriveRefused(t, h.checkRedshiftRetryGate(context.Background(), failed), "at least 24h ago")
+}
+
+func TestCheckRedshiftRetryGate_InsideSettleRetryAfter(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	// 9m59.5s since update leaves 500ms to wait, which must round up to 1s.
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: 10*time.Minute - 500*time.Millisecond, SinceStamp: ageOf(time.Hour)})
+
+	err := h.checkRedshiftRetryGate(context.Background(), failed)
+
+	ce, ok := IsClientError(err)
+	require.True(t, ok, "got %v", err)
+	assert.Equal(t, 409, ce.code)
+	assert.Equal(t, 1, ce.Details()["retry_after_seconds"])
+	assert.NotContains(t, ce.Details(), "redrive_unsafe", "a settle wait is temporary")
+	assert.Equal(t, "1", ce.headers["Retry-After"])
+}
+
+func TestCheckRedshiftRetryGate_AfterSettleAllowed(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: 10 * time.Minute, SinceStamp: ageOf(time.Hour)})
+
+	assert.NoError(t, h.checkRedshiftRetryGate(context.Background(), failed))
+}
+
+func TestCheckRedshiftRetryGate_NoSubmissionAttempt0Skipped(t *testing.T) {
+	failed := redshiftFailedRow(0, true)
+	h := gateWithClock(t, failed, config.RedshiftRetryClock{SinceUpdate: 0, SinceStamp: nil})
+
+	assert.NoError(t, h.checkRedshiftRetryGate(context.Background(), failed))
+}
+
+func TestCheckRedshiftRetryGate_NonRedshiftNeverReadsClock(t *testing.T) {
+	for _, svc := range []string{"ec2", "rds"} {
+		failed := redriveFailedRow(redshiftRetryExecID, "aws", svc)
+		store := new(MockConfigStore)
+		h := &Handler{config: store}
+
+		assert.NoError(t, h.checkRedshiftRetryGate(context.Background(), failed), svc)
+		store.AssertNotCalled(t, "GetRedshiftRetryClock", mock.Anything, mock.Anything)
+	}
+}
+
+func TestCheckRedshiftRetryGate_ClockErrorFailsClosed(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	store := new(MockConfigStore)
+	store.On("GetRedshiftRetryClock", mock.Anything, failed.ExecutionID).Return(config.RedshiftRetryClock{}, assert.AnError)
+	h := &Handler{config: store}
+
+	err := h.checkRedshiftRetryGate(context.Background(), failed)
+
+	require.ErrorIs(t, err, assert.AnError)
+	_, isClient := IsClientError(err)
+	assert.False(t, isClient, "a store failure is a 500, not a refusal the UI would render as permanent")
+}
+
+// Until the old guard is lifted it refuses every Redshift Retry first, so the
+// new gate must not even read the clock.
+func TestLoadAndValidateRetry_OldRedriveRefusalStillWinsForRedshift(t *testing.T) {
+	failed := redshiftFailedRow(1, false)
+	session := &Session{UserID: retryCallerID, Email: "operator@example.com"}
+	h, store, _ := buildSessionRetryHandler(t, failed, session, false, true)
+
+	_, _, err := h.loadAndValidateRetryRequest(context.Background(), sessionRetryReqWithForce(), failed.ExecutionID)
+
+	assertRedriveRefused(t, err, "Check EC2 Reserved Instances or Redshift Reserved Nodes")
+	store.AssertNotCalled(t, "GetRedshiftRetryClock", mock.Anything, mock.Anything)
+}
+
+func persistedSuccessor(t *testing.T, failed *config.PurchaseExecution) *config.PurchaseExecution {
+	t.Helper()
+	session := &Session{UserID: retryCallerID, Email: "operator@example.com"}
+	store := new(MockConfigStore)
+	t.Cleanup(func() { store.AssertExpectations(t) })
+	store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{}, nil).Maybe()
+	h := &Handler{config: store}
+	var saved *config.PurchaseExecution
+	store.On("SavePurchaseExecution", mock.Anything, mock.AnythingOfType("*config.PurchaseExecution")).
+		Run(func(args mock.Arguments) { snap := *args.Get(1).(*config.PurchaseExecution); saved = &snap }).
+		Return(nil)
+	store.On("LinkRetryExecutionAtomic", mock.Anything, mock.Anything, failed.ExecutionID, mock.AnythingOfType("string")).Return(true, nil)
+
+	_, _, err := h.persistRetryExecution(context.Background(), failed, session, 0, 0)
+
+	require.NoError(t, err)
+	require.NotNil(t, saved)
+	return saved
+}
+
+func TestPersistRetryExecution_CopiesFirstDispatchAt(t *testing.T) {
+	stamp := time.Now().Add(-time.Hour).UTC().Truncate(time.Microsecond)
+	failed := redshiftFailedRow(1, false)
+	failed.FirstDispatchAt = &stamp
+
+	successor := persistedSuccessor(t, failed)
+
+	require.NotNil(t, successor.FirstDispatchAt)
+	assert.True(t, stamp.Equal(*successor.FirstDispatchAt), "the successor keeps the lineage's original stamp, not a new one")
+	assert.NotSame(t, failed.FirstDispatchAt, successor.FirstDispatchAt, "the successor must not alias the predecessor's pointer")
+}
+
+func TestPersistRetryExecution_NoSubmissionAttempt0LeavesStampNil(t *testing.T) {
+	stamp := time.Now().Add(-time.Hour)
+	failed := redshiftFailedRow(0, true)
+	failed.FirstDispatchAt = &stamp
+
+	assert.Nil(t, persistedSuccessor(t, failed).FirstDispatchAt)
+}
+
+func TestPersistRetryExecution_StamplessPredecessorStaysStampless(t *testing.T) {
+	assert.Nil(t, persistedSuccessor(t, redshiftFailedRow(1, false)).FirstDispatchAt)
+}
diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go
index cae5f0833..b9f80a5d4 100644
--- a/internal/config/interfaces.go
+++ b/internal/config/interfaces.go
@@ -827,6 +827,10 @@ type StoreInterface interface {
 	// not race (see the implementation for the predicate).
 	FindOpenRedshiftLineageConflictsTx(ctx context.Context, tx pgx.Tx, selfID, family string, tuples []RedshiftTuple, accountID *string, horizon time.Duration, stamp time.Time, window time.Duration) ([]RedshiftLineageConflict, error)
 
+	// GetRedshiftRetryClock returns how long ago an execution was last written
+	// and first dispatched, both measured on the database clock.
+	GetRedshiftRetryClock(ctx context.Context, executionID string) (RedshiftRetryClock, error)
+
 	// ListRecentSubmitsTx takes a per-creator transaction-scoped lock and
 	// returns the creator's web-submitted executions scheduled at or after
 	// since that are pending, in flight or bought. executePurchase's
diff --git a/internal/config/store_postgres_redshift_retry.go b/internal/config/store_postgres_redshift_retry.go
new file mode 100644
index 000000000..124051ea0
--- /dev/null
+++ b/internal/config/store_postgres_redshift_retry.go
@@ -0,0 +1,46 @@
+package config
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// RedshiftRetryClock is a row's age measured on the database clock.
+// SinceStamp is nil when the row has no first_dispatch_at, so a missing stamp
+// never reads as fresh.
+type RedshiftRetryClock struct {
+	SinceUpdate time.Duration
+	SinceStamp  *time.Duration
+}
+
+// GetRedshiftRetryClock returns how long ago the row was last written and
+// first dispatched, both computed from now() in SQL so app clock skew cannot
+// shorten the settle wait or the horizon.
+func (s *PostgresStore) GetRedshiftRetryClock(ctx context.Context, executionID string) (RedshiftRetryClock, error) {
+	var sinceUpdate float64
+	var sinceStamp *float64
+	err := s.db.QueryRow(ctx, `
+		SELECT EXTRACT(EPOCH FROM now() - updated_at)::float8,
+		       EXTRACT(EPOCH FROM now() - first_dispatch_at)::float8
+		FROM purchase_executions WHERE execution_id = $1`, executionID).Scan(&sinceUpdate, &sinceStamp)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return RedshiftRetryClock{}, fmt.Errorf("%w: execution %s", ErrNotFound, executionID)
+	}
+	if err != nil {
+		return RedshiftRetryClock{}, fmt.Errorf("read redshift retry clock of execution %s: %w", executionID, err)
+	}
+	clock := RedshiftRetryClock{SinceUpdate: secondsToDuration(sinceUpdate)}
+	if sinceStamp != nil {
+		d := secondsToDuration(*sinceStamp)
+		clock.SinceStamp = &d
+	}
+	return clock, nil
+}
+
+func secondsToDuration(secs float64) time.Duration {
+	return time.Duration(secs * float64(time.Second))
+}
diff --git a/internal/config/store_postgres_redshift_retry_integration_test.go b/internal/config/store_postgres_redshift_retry_integration_test.go
new file mode 100644
index 000000000..c51154c6e
--- /dev/null
+++ b/internal/config/store_postgres_redshift_retry_integration_test.go
@@ -0,0 +1,51 @@
+//go:build integration
+// +build integration
+
+package config
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// The ages come from the database clock. The 000128 trigger overwrites updated_at
+// on UPDATE, so the aged row is INSERTed with explicit timestamps.
+func TestGetRedshiftRetryClock_UsesDBNow(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	var id string
+	require.NoError(t, store.db.QueryRow(ctx, `
+		INSERT INTO purchase_executions (status, step_number, scheduled_date, first_dispatch_at, updated_at)
+		VALUES ('failed', 1, now(), now() - interval '3 hours', now() - interval '7 minutes')
+		RETURNING execution_id::text`).Scan(&id))
+
+	clock, err := store.GetRedshiftRetryClock(ctx, id)
+	require.NoError(t, err)
+	assert.InDelta(t, (7 * time.Minute).Seconds(), clock.SinceUpdate.Seconds(), 30)
+	require.NotNil(t, clock.SinceStamp)
+	assert.InDelta(t, (3 * time.Hour).Seconds(), clock.SinceStamp.Seconds(), 30)
+}
+
+func TestGetRedshiftRetryClock_NullStampIsNil(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	exec := newNoSubmissionExec(ctx, t, store, "failed")
+
+	clock, err := store.GetRedshiftRetryClock(ctx, exec.ExecutionID)
+	require.NoError(t, err)
+	assert.Nil(t, clock.SinceStamp, "a NULL stamp must not read as fresh")
+	assert.GreaterOrEqual(t, clock.SinceUpdate, time.Duration(0))
+}
+
+func TestGetRedshiftRetryClock_UnknownExecutionIsNotFound(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+
+	_, err := store.GetRedshiftRetryClock(ctx, "00000000-0000-0000-0000-000000000000")
+	assert.ErrorIs(t, err, ErrNotFound)
+}
diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go
index c83968e59..182e8e0e5 100644
--- a/internal/mocks/stores.go
+++ b/internal/mocks/stores.go
@@ -589,6 +589,17 @@ func (m *MockConfigStore) FindOpenRedshiftLineageConflictsTx(ctx context.Context
 	return res, args.Error(1)
 }
 
+// GetRedshiftRetryClock mocks the Redshift retry clock read (no default: testify panics if unexpected).
+func (m *MockConfigStore) GetRedshiftRetryClock(ctx context.Context, executionID string) (config.RedshiftRetryClock, error) {
+	m.record("GetRedshiftRetryClock", ctx, executionID)
+	args := m.Called(ctx, executionID)
+	clock, ok := args.Get(0).(config.RedshiftRetryClock)
+	if !ok {
+		m.reporter.wrongReturnType("GetRedshiftRetryClock", "config.RedshiftRetryClock", args.Get(0))
+	}
+	return clock, args.Error(1)
+}
+
 // TransitionExecutionStatus mocks the TransitionExecutionStatus operation.
 func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*config.PurchaseExecution, error) {
 	m.record("TransitionExecutionStatus", ctx, executionID, fromStatuses, toStatus, actor)
diff --git a/internal/purchase/idempotency_key_writers_test.go b/internal/purchase/idempotency_key_writers_test.go
new file mode 100644
index 000000000..ccae86ed7
--- /dev/null
+++ b/internal/purchase/idempotency_key_writers_test.go
@@ -0,0 +1,323 @@
+package purchase
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const idempotencyKeySettleContract = "A purchase lineage's IdempotencyKey is minted once and copied verbatim; the provider token and the go#107 " +
+	"attempt journal's settle contract key on it. If this writer is intended, review it against that contract and then update the pinned set."
+
+// pinnedIdempotencyKeyWriters is every function that writes PurchaseExecution.IdempotencyKey.
+var pinnedIdempotencyKeyWriters = []string{
+	"internal/api.Handler.persistRetryExecution",
+	"internal/config.PostgresStore.SavePurchaseExecutionTx",
+	"internal/config.scanExecutionRows",
+	"internal/purchase.Manager.executeForAccount",
+	"internal/purchase.Manager.writeAttached",
+}
+
+func isIdempotencyKeySelector(e ast.Expr) bool {
+	for {
+		p, ok := e.(*ast.ParenExpr)
+		if !ok {
+			break
+		}
+		e = p.X
+	}
+	sel, ok := e.(*ast.SelectorExpr)
+	return ok && sel.Sel.Name == "IdempotencyKey"
+}
+
+func receiverTypeName(e ast.Expr) string {
+	switch t := e.(type) {
+	case *ast.StarExpr:
+		return receiverTypeName(t.X)
+	case *ast.IndexExpr:
+		return receiverTypeName(t.X)
+	case *ast.IndexListExpr:
+		return receiverTypeName(t.X)
+	case *ast.Ident:
+		return t.Name
+	}
+	return "?"
+}
+
+func declLabel(dir string, d ast.Decl) string {
+	fn, ok := d.(*ast.FuncDecl)
+	if !ok {
+		return dir + ".<package-level>"
+	}
+	if fn.Recv != nil && len(fn.Recv.List) == 1 {
+		return dir + "." + receiverTypeName(fn.Recv.List[0].Type) + "." + fn.Name.Name
+	}
+	return dir + "." + fn.Name.Name
+}
+
+// idempotencyKeyWriters lists the declarations that assign, take the address of, or
+// key a literal with IdempotencyKey, by identifier. Blind: whole-struct copies, SQL from non-literal pieces.
+func idempotencyKeyWriters(f *ast.File, dir string) []string {
+	var out []string
+	for _, d := range f.Decls {
+		label := declLabel(dir, d)
+		ast.Inspect(d, func(n ast.Node) bool {
+			switch x := n.(type) {
+			case *ast.AssignStmt:
+				for _, lhs := range x.Lhs {
+					if isIdempotencyKeySelector(lhs) {
+						out = append(out, label)
+					}
+				}
+			case *ast.UnaryExpr:
+				if x.Op == token.AND && isIdempotencyKeySelector(x.X) {
+					out = append(out, label)
+				}
+			case *ast.KeyValueExpr:
+				if id, ok := x.Key.(*ast.Ident); ok && id.Name == "IdempotencyKey" {
+					out = append(out, label)
+				}
+			}
+			return true
+		})
+	}
+	return out
+}
+
+var (
+	insertPurchaseExecutions = regexp.MustCompile(`(?is)\bINSERT\s+INTO\s+purchase_executions\b`)
+	updatePurchaseExecutions = regexp.MustCompile(`(?is)\bUPDATE\s+purchase_executions\b.*?\bSET\b(.*?)(?:\bWHERE\b|\bRETURNING\b|;|$)`)
+	upsertSet                = regexp.MustCompile(`(?is)\bDO\s+UPDATE\s+SET\b(.*?)(?:\bWHERE\b|\bRETURNING\b|;|$)`)
+	idempotencyKeyColumn     = regexp.MustCompile(`(?i)\bidempotency_key\b`)
+)
+
+// columnList returns the balanced "(...)" after an INSERT target and the text
+// following it; ok is false when the list is missing its close.
+func columnList(rest string) (cols, tail string, ok bool) {
+	depth := 0
+	for i, r := range rest {
+		switch r {
+		case '(':
+			depth++
+		case ')':
+			if depth--; depth == 0 {
+				return rest[:i+1], rest[i+1:], true
+			}
+		}
+	}
+	return "", "", false
+}
+
+// sqlIdempotencyKeyWriters counts purchase_executions INSERTs carrying idempotency_key
+// (no column list counts) and UPDATEs or upserts that SET it; dynamic counts unverifiable INSERTs.
+func sqlIdempotencyKeyWriters(sql string) (inserts, updates, dynamic int) {
+	for _, loc := range insertPurchaseExecutions.FindAllStringIndex(sql, -1) {
+		rest := strings.TrimSpace(sql[loc[1]:])
+		if !strings.HasPrefix(rest, "(") {
+			inserts++
+			continue
+		}
+		cols, tail, ok := columnList(rest)
+		if !ok || strings.Contains(cols, "%") || strings.TrimSpace(cols[1:len(cols)-1]) == "" {
+			dynamic++
+			continue
+		}
+		if idempotencyKeyColumn.MatchString(cols) {
+			inserts++
+		}
+		if end := strings.Index(tail, ";"); end >= 0 {
+			tail = tail[:end]
+		}
+		for _, m := range upsertSet.FindAllStringSubmatch(tail, -1) {
+			if idempotencyKeyColumn.MatchString(m[1]) {
+				updates++
+			}
+		}
+	}
+	for _, m := range updatePurchaseExecutions.FindAllStringSubmatch(sql, -1) {
+		if idempotencyKeyColumn.MatchString(m[1]) {
+			updates++
+		}
+	}
+	return inserts, updates, dynamic
+}
+
+func sqlLiteralWriters(f *ast.File) (inserts, updates, dynamic int) {
+	ast.Inspect(f, func(n ast.Node) bool {
+		lit, ok := n.(*ast.BasicLit)
+		if !ok || lit.Kind != token.STRING {
+			return true
+		}
+		if s, err := strconv.Unquote(lit.Value); err == nil {
+			i, u, d := sqlIdempotencyKeyWriters(s)
+			inserts += i
+			updates += u
+			dynamic += d
+		}
+		return true
+	})
+	return inserts, updates, dynamic
+}
+
+func moduleRoot(t *testing.T) string {
+	t.Helper()
+	dir, err := os.Getwd()
+	require.NoError(t, err)
+	for {
+		if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil {
+			return dir
+		}
+		parent := filepath.Dir(dir)
+		require.NotEqual(t, dir, parent, "go.mod not found above the test directory")
+		dir = parent
+	}
+}
+
+func TestIdempotencyKeyCopyHasSingleWriter(t *testing.T) {
+	root := moduleRoot(t)
+	fset := token.NewFileSet()
+	writers := map[string]bool{}
+	var inserts, updates, dynamic, files int
+
+	require.NoError(t, filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			name := d.Name()
+			if path != root && (strings.HasPrefix(name, ".") || name == "node_modules" || name == "vendor" || name == "testdata") {
+				return filepath.SkipDir
+			}
+			if _, statErr := os.Stat(filepath.Join(path, "go.mod")); path != root && statErr == nil {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+		if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+			return nil
+		}
+		f, parseErr := parser.ParseFile(fset, path, nil, parser.SkipObjectResolution)
+		if parseErr != nil {
+			return parseErr
+		}
+		files++
+		rel, relErr := filepath.Rel(root, filepath.Dir(path))
+		if relErr != nil {
+			return relErr
+		}
+		for _, w := range idempotencyKeyWriters(f, filepath.ToSlash(rel)) {
+			writers[w] = true
+		}
+		i, u, dyn := sqlLiteralWriters(f)
+		inserts += i
+		updates += u
+		dynamic += dyn
+		return nil
+	}))
+
+	require.Greater(t, files, 100, "the scan must cover the module, not a subtree")
+	got := make([]string, 0, len(writers))
+	for w := range writers {
+		got = append(got, w)
+	}
+	sort.Strings(got)
+	want := append([]string(nil), pinnedIdempotencyKeyWriters...)
+	sort.Strings(want)
+	assert.Equal(t, want, got, idempotencyKeySettleContract)
+	assert.Zero(t, dynamic, "cannot verify idempotency_key writers in a dynamically built purchase_executions INSERT. "+idempotencyKeySettleContract)
+	assert.Equal(t, 1, inserts, "exactly one INSERT INTO purchase_executions may carry idempotency_key. "+idempotencyKeySettleContract)
+	assert.Equal(t, 0, updates, "no UPDATE purchase_executions may SET idempotency_key. "+idempotencyKeySettleContract)
+}
+
+func parseFixture(t *testing.T, src string) *ast.File {
+	t.Helper()
+	f, err := parser.ParseFile(token.NewFileSet(), "fixture.go", src, parser.SkipObjectResolution)
+	require.NoError(t, err)
+	return f
+}
+
+func TestIdempotencyKeyWriterScan_Detects(t *testing.T) {
+	tests := []struct {
+		name string
+		src  string
+		want []string
+	}{
+		{"selector assignment", `package p
+func f(e *X) { e.IdempotencyKey = "a" }`, []string{"d.f"}},
+		{"composite literal key", `package p
+func g() { _ = X{IdempotencyKey: "a"} }`, []string{"d.g"}},
+		{"aliased import and method receiver", `package p
+import cfg "example.com/config"
+func (m *Manager) h(k string) { var e cfg.PurchaseExecution; e.IdempotencyKey = k; _ = &cfg.PurchaseExecution{IdempotencyKey: k} }`, []string{"d.Manager.h"}},
+		{"address of field", `package p
+func (s Store) scan(e *X) { _ = row.Scan(&e.IdempotencyKey) }`, []string{"d.Store.scan"}},
+		{"parenthesised assignment", `package p
+func p1(e *X) { (e.IdempotencyKey) = "a" }`, []string{"d.p1"}},
+		{"package level literal", `package p
+var def = X{IdempotencyKey: "a"}`, []string{"d.<package-level>"}},
+		{"reads and declarations are not writers", `package p
+type X struct{ IdempotencyKey string }
+func r(e *X) string { k := e.IdempotencyKey; use(e.IdempotencyKey); return k }`, nil},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.ElementsMatch(t, tc.want, dedupe(idempotencyKeyWriters(parseFixture(t, tc.src), "d")))
+		})
+	}
+}
+
+func dedupe(in []string) []string {
+	seen := map[string]bool{}
+	var out []string
+	for _, s := range in {
+		if !seen[s] {
+			seen[s] = true
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+func TestIdempotencyKeySQLScan_Detects(t *testing.T) {
+	tests := []struct {
+		name                      string
+		sql                       string
+		inserts, updates, dynamic int
+	}{
+		{"insert with column", `INSERT INTO purchase_executions (execution_id, idempotency_key) VALUES ($1, $2)`, 1, 0, 0},
+		{"insert without the column", `INSERT INTO purchase_executions (execution_id, status) VALUES ($1, $2)`, 0, 0, 0},
+		{"insert select copy", "insert into purchase_executions (idempotency_key, status)\nSELECT idempotency_key, 'x' FROM purchase_executions", 1, 0, 0},
+		{"insert with no column list", `INSERT INTO purchase_executions SELECT * FROM purchase_executions`, 1, 0, 0},
+		{"update sets it", `UPDATE purchase_executions SET status = $2, idempotency_key = $3 WHERE execution_id = $1`, 0, 1, 0},
+		{"update only filters on it", `UPDATE purchase_executions SET status = $2 WHERE idempotency_key = $1`, 0, 0, 0},
+		{"upsert update clause is not a separate update", `INSERT INTO purchase_executions (execution_id, idempotency_key) VALUES ($1, $2) ON CONFLICT (execution_id) DO UPDATE SET status = EXCLUDED.status`, 1, 0, 0},
+		{"other table is out of scope", `INSERT INTO ri_exchange_idempotency (idempotency_key, claimed_at) VALUES ($1, now())`, 0, 0, 0},
+		{"similarly named table is out of scope", `INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape) VALUES ($1, $2)`, 0, 0, 0},
+		{"upsert rewrites the key from EXCLUDED", `INSERT INTO purchase_executions (execution_id, status) VALUES ($1, $2) ON CONFLICT (execution_id) DO UPDATE SET idempotency_key = EXCLUDED.idempotency_key`, 0, 1, 0},
+		{"upsert rewrites the key from a parameter", `INSERT INTO purchase_executions (execution_id) VALUES ($1) ON CONFLICT (execution_id) DO UPDATE SET status = $2, idempotency_key = $3 WHERE true`, 0, 1, 0},
+		{"update row constructor", `UPDATE purchase_executions SET (idempotency_key, status) = ($2, $3) WHERE execution_id = $1`, 0, 1, 0},
+		{"update key set after a FROM inside parentheses", `UPDATE purchase_executions SET status = (SELECT s FROM t), idempotency_key = $2 WHERE execution_id = $1`, 0, 1, 0},
+		{"insert column list built with Sprintf", `INSERT INTO purchase_executions (%s) VALUES (%s)`, 0, 0, 1},
+		{"insert column list unclosed by concatenation", `INSERT INTO purchase_executions (execution_id, `, 0, 0, 1},
+		{"insert column list empty before concatenation", `INSERT INTO purchase_executions (`, 0, 0, 1},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			i, u, d := sqlIdempotencyKeyWriters(tc.sql)
+			assert.Equal(t, tc.dynamic, d, "dynamic")
+			assert.Equal(t, tc.inserts, i, "inserts")
+			assert.Equal(t, tc.updates, u, "updates")
+		})
+	}
+}
diff --git a/internal/purchase/redshift_retry.go b/internal/purchase/redshift_retry.go
new file mode 100644
index 000000000..f73641d9a
--- /dev/null
+++ b/internal/purchase/redshift_retry.go
@@ -0,0 +1,56 @@
+package purchase
+
+import (
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+)
+
+// redshiftRetrySettle is how long after a failed Redshift row's last write a
+// Retry waits, so a purchase AWS accepted late is visible to the dispatch dedupe.
+const redshiftRetrySettle = 10 * time.Minute
+
+// Fixed id-free texts. A stamp-less row is not necessarily old (a reaped
+// never-sent row has none either), so no text says so.
+const (
+	redshiftRetryNoStampRefusal = "this Redshift purchase has no first dispatch time recorded, so a retry could buy a second reservation; check the account in AWS and start a new purchase"
+	redshiftRetryHorizonRefusal = "this Redshift purchase was first dispatched at least 24h ago, so a retry could buy a second reservation; check the account in AWS and start a new purchase"
+)
+
+// IsRedshiftExecution reports whether exec has a selected Redshift recommendation.
+func IsRedshiftExecution(exec *config.PurchaseExecution) bool {
+	return hasRedshiftRec(exec)
+}
+
+// NeverSubmitted reports a first-attempt row proven to have sent nothing.
+func NeverSubmitted(exec *config.PurchaseExecution) bool {
+	return exec.NoSubmission && exec.RetryAttemptN == 0
+}
+
+// RedshiftRetryPermanentRefusal returns the reason a failed Redshift execution
+// can never be retried, or "". sinceStamp is the age of its first dispatch on
+// the database clock, nil when the row has no stamp.
+func RedshiftRetryPermanentRefusal(exec *config.PurchaseExecution, sinceStamp *time.Duration) string {
+	if !hasRedshiftRec(exec) {
+		return ""
+	}
+	if NeverSubmitted(exec) {
+		return "" // its successor mints a fresh stamp, so the old one is irrelevant
+	}
+	if sinceStamp == nil {
+		return redshiftRetryNoStampRefusal
+	}
+	if *sinceStamp >= redshiftRedriveHorizon {
+		return redshiftRetryHorizonRefusal
+	}
+	return ""
+}
+
+// RedshiftRetryWait returns how much longer a Retry must wait for sinceUpdate
+// (age of the row's last write on the database clock) to reach the settle time.
+func RedshiftRetryWait(exec *config.PurchaseExecution, sinceUpdate time.Duration) time.Duration {
+	if !hasRedshiftRec(exec) || NeverSubmitted(exec) || sinceUpdate >= redshiftRetrySettle {
+		return 0
+	}
+	return redshiftRetrySettle - sinceUpdate
+}
diff --git a/internal/purchase/redshift_retry_test.go b/internal/purchase/redshift_retry_test.go
new file mode 100644
index 000000000..a74447251
--- /dev/null
+++ b/internal/purchase/redshift_retry_test.go
@@ -0,0 +1,80 @@
+package purchase
+
+import (
+	"testing"
+	"time"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
+	"github.com/stretchr/testify/assert"
+)
+
+func retryRow(attempt int, noSubmission bool, recs ...config.RecommendationRecord) *config.PurchaseExecution {
+	return &config.PurchaseExecution{
+		ExecutionID: "rs-failed", Status: "failed", RetryAttemptN: attempt, NoSubmission: noSubmission, Recommendations: recs,
+	}
+}
+
+func durPtr(d time.Duration) *time.Duration { return &d }
+
+func TestRedshiftRetryPermanentRefusal(t *testing.T) {
+	rs := redshiftScopedRec("acct-A")
+	ec2 := scopedTestRec("acct-A")
+	ec2.Service = "ec2"
+	h := redshiftRedriveHorizon
+	tests := []struct {
+		name  string
+		exec  *config.PurchaseExecution
+		since *time.Duration
+		want  string
+	}{
+		{"no stamp refused", retryRow(0, false, rs), nil, redshiftRetryNoStampRefusal},
+		{"no stamp attempt 1 no_submission refused", retryRow(1, true, rs), nil, redshiftRetryNoStampRefusal},
+		{"no stamp attempt 0 no_submission skipped", retryRow(0, true, rs), nil, ""},
+		{"fresh stamp allowed", retryRow(0, false, rs), durPtr(0), ""},
+		{"just under horizon allowed", retryRow(1, false, rs), durPtr(h - time.Nanosecond), ""},
+		{"exactly horizon refused", retryRow(1, false, rs), durPtr(h), redshiftRetryHorizonRefusal},
+		{"past horizon refused", retryRow(1, false, rs), durPtr(h + time.Second), redshiftRetryHorizonRefusal},
+		// A never-submitted attempt 0 is exempt: its successor mints a fresh stamp.
+		{"past horizon attempt 0 no_submission exempt", retryRow(0, true, rs), durPtr(h), ""},
+		{"non-redshift never refused", retryRow(0, false, ec2), nil, ""},
+		{"unselected redshift rec never refused", retryRow(0, false, func() config.RecommendationRecord { r := rs; r.Selected = false; return r }()), nil, ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, RedshiftRetryPermanentRefusal(tc.exec, tc.since))
+		})
+	}
+}
+
+func TestRedshiftRetryWait(t *testing.T) {
+	rs := redshiftScopedRec("acct-A")
+	ec2 := scopedTestRec("acct-A")
+	ec2.Service = "ec2"
+	s := redshiftRetrySettle
+	tests := []struct {
+		name  string
+		exec  *config.PurchaseExecution
+		since time.Duration
+		want  time.Duration
+	}{
+		{"just written waits full settle", retryRow(1, false, rs), 0, s},
+		{"midway", retryRow(1, false, rs), 4 * time.Minute, 6 * time.Minute},
+		{"one nanosecond short", retryRow(1, false, rs), s - time.Nanosecond, time.Nanosecond},
+		{"exactly settled", retryRow(1, false, rs), s, 0},
+		{"long settled", retryRow(1, false, rs), time.Hour, 0},
+		{"never submitted attempt 0 skips wait", retryRow(0, true, rs), 0, 0},
+		{"attempt 1 no_submission still waits", retryRow(1, true, rs), 0, s},
+		{"non-redshift never waits", retryRow(1, false, ec2), 0, 0},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.want, RedshiftRetryWait(tc.exec, tc.since))
+		})
+	}
+}
+
+func TestRedshiftRetryRefusalsContainNoLegacyWording(t *testing.T) {
+	for _, text := range []string{redshiftRetryNoStampRefusal, redshiftRetryHorizonRefusal} {
+		assert.NotContains(t, text, "legacy")
+	}
+}