From 605b447756112ae85dc73f315acf94675fbf3728 Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 01:53:47 -0700
Subject: [PATCH 1/7] fix(purchases): close double-buy holes in idempotency,
 atomic claim, and partial-success labelling

Three interlocking money-path defects that could cause DUPLICATE real cloud
commitment purchases:

#1012 - idempotency identity was tied to the mutable execution ID, so retry
(fresh UUID) and multi-account fan-out (per-account UUID) derived a DIFFERENT
provider token than the original attempt, defeating provider-level dedupe.
  - Add a stable idempotency_key column (migration 000066), generated once at
    first creation, INSERT-only (never rewritten by ON CONFLICT).
  - Derive the per-rec token from idempotencyLineageKey(exec) (the key, with an
    ExecutionID fallback for legacy NULL rows) instead of exec.ExecutionID.
  - Copy the key verbatim onto retry successors; seed each multi-account row's
    key from rootKey+accountID (deterministic, not uuid.New()).

#1013 - the atomic row-claim that the sync approve path has was missing on the
SQS execute_purchase and cron paths, so at-least-once SQS redelivery and
overlapping cron ticks could double-execute the same row.
  - Add claimAndExecute: CAS [approved,pending,notified] -> running; only the
    winner proceeds, a lost CAS is a benign ack/skip.
  - Route handleExecutePurchase and ProcessScheduledPurchases through it; add a
    positive-allowlist guard on the cron loop (M2).

#1014 - multi-account partial success was joined into an opaque error string
that finalizeExecution classified as "failed", and an SQS handler returning it
invited redelivery -> re-execution -> double-buy.
  - executeMultiAccount returns a typed *multiAccountPartialError (>=1 account
    committed) vs errAllAccountsFailed (nothing committed).
  - finalizeExecution maps the partial sentinel to partially_completed, never
    failed; the root row is now always saved with its aggregate status (H2),
    fixing the stranded-"running" root the new claim would otherwise leave.
  - SQS/cron callers ack (no redeliver) when >=1 account committed.

Regression tests replicate the real scenarios and fail on pre-fix code:
retry/fan-out reuse the same token; a redelivered SQS message executes the
cloud purchase exactly once; a multi-account partial run is acked and recorded
partially_completed (not failed).

Closes #1012
Closes #1013
Closes #1014
---
 internal/api/handler_purchases.go             |  12 +
 internal/config/store_postgres.go             |  48 ++-
 .../config/store_postgres_pgxmock_test.go     |   6 +
 internal/config/types.go                      |  11 +
 ...rchase_executions_idempotency_key.down.sql |   3 +
 ...purchase_executions_idempotency_key.up.sql |  24 ++
 internal/purchase/coverage_extra_test.go      |  17 +-
 internal/purchase/execution.go                | 114 +++++-
 internal/purchase/execution_test.go           |   3 +-
 internal/purchase/manager.go                  | 166 ++++++--
 internal/purchase/manager_test.go             |   9 +
 internal/purchase/messages.go                 |  42 +-
 internal/purchase/messages_test.go            |   8 +
 .../purchase/money_path_regression_test.go    | 358 ++++++++++++++++++
 14 files changed, 738 insertions(+), 83 deletions(-)
 create mode 100644 internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.down.sql
 create mode 100644 internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.up.sql
 create mode 100644 internal/purchase/money_path_regression_test.go

diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go
index 185147d8e..72c1ae4e1 100644
--- a/internal/api/handler_purchases.go
+++ b/internal/api/handler_purchases.go
@@ -1064,12 +1064,24 @@ func (h *Handler) persistRetryExecution(ctx context.Context, failedExec *config.
 	tokenExpiresAt := time.Now().Add(config.ApprovalTokenTTL)
 
 	newExecutionID := uuid.New().String()
+	// Carry the predecessor's stable idempotency lineage key onto the retry
+	// successor VERBATIM (issue #1012) so the per-rec provider token is
+	// reproduced and a re-drive of a "failed" execution whose commitment
+	// actually landed short-circuits at the provider instead of double-buying.
+	// Legacy predecessors (persisted before migration 000066) carry no key —
+	// seed the successor's key from the predecessor's ExecutionID, which is what
+	// the original attempt's token derived from, so the match still holds.
+	idempotencyKey := failedExec.IdempotencyKey
+	if idempotencyKey == "" {
+		idempotencyKey = failedExec.ExecutionID
+	}
 	// PlanID + StepNumber propagate from the predecessor so a retried
 	// planned execution stays attributed to its plan + ramp step (CR
 	// #168 review). For ad-hoc executions PlanID is "" and StepNumber
 	// is 0, so propagation is a no-op for the non-plan case.
 	newExecution := &config.PurchaseExecution{
 		ExecutionID:            newExecutionID,
+		IdempotencyKey:         idempotencyKey,
 		PlanID:                 failedExec.PlanID,
 		StepNumber:             failedExec.StepNumber,
 		Status:                 "pending",
diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go
index b489e70d6..d599481cc 100644
--- a/internal/config/store_postgres.go
+++ b/internal/config/store_postgres.go
@@ -723,6 +723,16 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 		execution.ExecutionID = uuid.New().String()
 	}
 
+	// Stamp a stable idempotency lineage key on first creation (issue #1012).
+	// Unlike ExecutionID this is generated once and then copied verbatim onto
+	// Retry successors / multi-account fan-out rows by the callers, so the
+	// derived provider token survives a re-drive. INSERT-only below (omitted
+	// from the ON CONFLICT DO UPDATE SET), so an upsert of an existing row
+	// never overwrites the key already persisted at creation.
+	if execution.IdempotencyKey == "" {
+		execution.IdempotencyKey = uuid.New().String()
+	}
+
 	// Marshal recommendations to JSONB
 	recommendationsJSON, err := json.Marshal(execution.Recommendations)
 	if err != nil {
@@ -744,8 +754,9 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 			cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 			created_by_user_id, retry_execution_id, retry_attempt_n,
 			approval_token_expires_at,
-			executed_by_user_id, executed_at, pre_approval_skip_reason
-		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25)
+			executed_by_user_id, executed_at, pre_approval_skip_reason,
+			idempotency_key
+		) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26)
 		ON CONFLICT (execution_id) DO UPDATE SET
 			status = $3,
 			notification_sent = $6,
@@ -815,6 +826,7 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 		execution.ExecutedByUserID,
 		execution.ExecutedAt,
 		execution.PreApprovalSkipReason,
+		execution.IdempotencyKey,
 	)
 
 	if err != nil {
@@ -838,7 +850,8 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution
 		          cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		          created_by_user_id, retry_execution_id, retry_attempt_n,
 		          approval_token_expires_at,
-		          executed_by_user_id, executed_at, pre_approval_skip_reason
+		          executed_by_user_id, executed_at, pre_approval_skip_reason,
+		          idempotency_key
 	`
 
 	records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses)
@@ -942,7 +955,8 @@ func (s *PostgresStore) GetExecutionsByStatuses(ctx context.Context, statuses []
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		ORDER BY scheduled_date DESC
@@ -1005,7 +1019,8 @@ func (s *PostgresStore) GetStaleApprovedExecutions(ctx context.Context, olderTha
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE status = 'approved' AND updated_at < NOW() - $1::interval
 	`
@@ -1046,7 +1061,8 @@ func (s *PostgresStore) ListStuckExecutions(ctx context.Context, statuses []stri
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		  AND updated_at < NOW() - $2::interval
@@ -1066,7 +1082,8 @@ func (s *PostgresStore) GetPendingExecutions(ctx context.Context) ([]PurchaseExe
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE status IN ('pending', 'notified')
 		  AND (expires_at IS NULL OR expires_at > NOW())
@@ -1089,7 +1106,8 @@ func (s *PostgresStore) GetPendingExecutionsTx(ctx context.Context, tx pgx.Tx) (
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE status IN ('pending', 'notified')
 		  AND (expires_at IS NULL OR expires_at > NOW())
@@ -1114,7 +1132,8 @@ func (s *PostgresStore) GetExecutionByID(ctx context.Context, executionID string
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE execution_id = $1
 	`
@@ -1140,7 +1159,8 @@ func (s *PostgresStore) GetExecutionByPlanAndDate(ctx context.Context, planID st
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
 		       approval_token_expires_at,
-		       executed_by_user_id, executed_at, pre_approval_skip_reason
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE plan_id = $1 AND scheduled_date = $2
 	`
@@ -1228,6 +1248,10 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 		// plan_id is nullable since migration 000033 (direct-execute
 		// rows from the Recommendations page have no originating plan).
 		var planID sql.NullString
+		// idempotency_key is NULL on rows created before migration 000066;
+		// leave exec.IdempotencyKey "" for those so the derivation falls back
+		// to ExecutionID (issue #1012).
+		var idempotencyKey sql.NullString
 
 		err := rows.Scan(
 			&planID,
@@ -1255,6 +1279,7 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 			&exec.ExecutedByUserID,
 			&executedAt,
 			&exec.PreApprovalSkipReason,
+			&idempotencyKey,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("failed to scan execution: %w", err)
@@ -1263,6 +1288,9 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 		if planID.Valid {
 			exec.PlanID = planID.String
 		}
+		if idempotencyKey.Valid {
+			exec.IdempotencyKey = idempotencyKey.String
+		}
 
 		// Unmarshal recommendations
 		if err := json.Unmarshal(recommendationsJSON, &exec.Recommendations); err != nil {
diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go
index 74c6e824e..7fc9802a3 100644
--- a/internal/config/store_postgres_pgxmock_test.go
+++ b/internal/config/store_postgres_pgxmock_test.go
@@ -436,6 +436,7 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) {
 		"created_by_user_id", "retry_execution_id", "retry_attempt_n",
 		"approval_token_expires_at",
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
+		"idempotency_key",
 	}
 	rows := pgxmock.NewRows(cols).AddRow(
 		"plan-1", "exec-1", "pending", 1, now,
@@ -445,6 +446,7 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) {
 		nil, nil, 0,
 		sql.NullTime{},
 		nil, sql.NullTime{}, nil,
+		nil, // idempotency_key (NULL: legacy-row scan path, migration 000066)
 	)
 	mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows)
 
@@ -478,6 +480,7 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) {
 		"created_by_user_id", "retry_execution_id", "retry_attempt_n",
 		"approval_token_expires_at",
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
+		"idempotency_key",
 	}
 	successorID := "exec-3"
 	rows := pgxmock.NewRows(cols).AddRow(
@@ -491,6 +494,7 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) {
 		nil, &successorID, 2,
 		sql.NullTime{},
 		nil, sql.NullTime{}, nil,
+		"idem-key-exec-2", // idempotency_key non-NULL: exercises the scan path
 	)
 	mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows)
 
@@ -1800,6 +1804,7 @@ func stuckExecRow(execID, status string, scheduled time.Time) []any {
 		nil, nil, 0,
 		sql.NullTime{},
 		nil, sql.NullTime{}, nil,
+		nil, // idempotency_key (NULL: legacy-row scan path, migration 000066)
 	}
 }
 
@@ -1812,6 +1817,7 @@ func stuckExecCols() []string {
 		"created_by_user_id", "retry_execution_id", "retry_attempt_n",
 		"approval_token_expires_at",
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
+		"idempotency_key",
 	}
 }
 
diff --git a/internal/config/types.go b/internal/config/types.go
index 9b3651ffd..2c04a4f76 100644
--- a/internal/config/types.go
+++ b/internal/config/types.go
@@ -286,6 +286,17 @@ type PurchaseExecution struct {
 	// string "direct-execute permission". NULL on every normal-flow row.
 	// Migration 000058.
 	PreApprovalSkipReason *string `json:"pre_approval_skip_reason,omitempty" dynamodbav:"pre_approval_skip_reason,omitempty"`
+	// IdempotencyKey is the stable lineage anchor the per-rec provider
+	// idempotency token is derived from (issue #1012). Unlike ExecutionID
+	// it is NOT regenerated on Retry or multi-account fan-out: it is
+	// generated once at first creation, copied verbatim onto every Retry
+	// successor, and combined with the account ID to seed each per-account
+	// fan-out row. This makes DeriveIdempotencyToken reproduce the same
+	// token across a strand-and-re-drive so the provider dedupes and the
+	// commitment is never bought twice. Empty on rows created before
+	// migration 000066 — the derivation falls back to ExecutionID for those
+	// (identical to the pre-fix behaviour for a single un-retried execution).
+	IdempotencyKey string `json:"idempotency_key,omitempty" dynamodbav:"idempotency_key,omitempty"`
 }
 
 // IsCancelable reports whether an execution may still be cancelled. Only the
diff --git a/internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.down.sql b/internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.down.sql
new file mode 100644
index 000000000..318cf8a09
--- /dev/null
+++ b/internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.down.sql
@@ -0,0 +1,3 @@
+-- Reverse 000066: drop the idempotency lineage key column.
+ALTER TABLE purchase_executions
+    DROP COLUMN IF EXISTS idempotency_key;
diff --git a/internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.up.sql b/internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.up.sql
new file mode 100644
index 000000000..e0b8c8367
--- /dev/null
+++ b/internal/database/postgres/migrations/000066_purchase_executions_idempotency_key.up.sql
@@ -0,0 +1,24 @@
+-- Stable idempotency lineage key for purchase executions (issue #1012).
+--
+-- The per-rec provider idempotency token was derived purely from the
+-- execution_id (DeriveIdempotencyToken(execution_id, rec_index)). But the
+-- execution_id is regenerated on Retry (a fresh UUID per successor row) and on
+-- multi-account fan-out (a fresh UUID per account row), so a re-drive of a
+-- "failed" execution whose commitment actually landed on the provider derived a
+-- DIFFERENT token and the provider's idempotency guard could not match,
+-- producing a duplicate purchase (double-buy).
+--
+-- idempotency_key is a stable lineage anchor that is:
+--   * generated once at first creation of an execution,
+--   * copied verbatim onto every Retry successor, and
+--   * combined with the account ID to seed each multi-account fan-out row,
+-- so the derived provider token survives both operations and a stranded-then-
+-- re-driven purchase reuses the identical token (AWS ClientToken / EC2 RI tag,
+-- Azure reservationOrderID GUID, GCP RequestId / commitment name).
+--
+-- Nullable TEXT: rows created before this migration read as NULL. The
+-- application falls back to the execution_id for the derivation on legacy rows
+-- (identical to the pre-fix behaviour for a single un-retried execution), so no
+-- backfill is required and old rows keep working.
+ALTER TABLE purchase_executions
+    ADD COLUMN IF NOT EXISTS idempotency_key TEXT;
diff --git a/internal/purchase/coverage_extra_test.go b/internal/purchase/coverage_extra_test.go
index c422e7dcf..ee88ca376 100644
--- a/internal/purchase/coverage_extra_test.go
+++ b/internal/purchase/coverage_extra_test.go
@@ -204,6 +204,11 @@ func TestHandleExecutePurchase_ApprovedStatus(t *testing.T) {
 	}
 
 	mockStore.On("GetExecutionByID", ctx, "exec-approved").Return(exec, nil)
+	// claimAndExecute CASes the row to "running" before executing (issue #1013).
+	runningExec := *exec
+	runningExec.Status = "running"
+	mockStore.On("TransitionExecutionStatus", ctx, "exec-approved",
+		[]string{"approved", "pending", "notified"}, "running").Return(&runningExec, nil)
 	mockStore.On("GetPurchasePlan", ctx, "plan-approved").Return(plan, nil)
 	mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
 	mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil)
@@ -281,6 +286,11 @@ func TestHandleExecutePurchase_SaveError(t *testing.T) {
 	}
 
 	mockStore.On("GetExecutionByID", ctx, "exec-save-err").Return(exec, nil)
+	// claimAndExecute CASes the row to "running" before executing (issue #1013).
+	runningExec := *exec
+	runningExec.Status = "running"
+	mockStore.On("TransitionExecutionStatus", ctx, "exec-save-err",
+		[]string{"approved", "pending", "notified"}, "running").Return(&runningExec, nil)
 	mockStore.On("GetPurchasePlan", ctx, "plan-save-err").Return(plan, nil)
 	mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(errors.New("save failed"))
 	mockSTS.On("GetCallerIdentity", ctx, mock.Anything).Return(nil, errors.New("sts error"))
@@ -301,9 +311,12 @@ func TestHandleExecutePurchase_SaveError(t *testing.T) {
 		ExecutionID: "exec-save-err",
 	}
 	err := manager.handleExecutePurchase(ctx, msg)
-	// Returns the save error (not the purchase error)
+	// The terminal-save failure now surfaces from executeAndFinalize as
+	// ErrAuditLoss (the row is stranded in "running"). The SQS handler returns
+	// it so the message is redelivered. A single-account provider failure with
+	// no committed recs is not multi-account-ackable, so it is returned.
 	assert.Error(t, err)
-	assert.Contains(t, err.Error(), "failed to save execution status")
+	assert.ErrorIs(t, err, config.ErrAuditLoss)
 }
 
 // Tests for ProcessMessage with approve and cancel happy paths.
diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go
index 013eb9187..0cdce9c93 100644
--- a/internal/purchase/execution.go
+++ b/internal/purchase/execution.go
@@ -127,34 +127,90 @@ func anyRecPurchased(recs []config.RecommendationRecord) bool {
 	return false
 }
 
+// multiAccountPartialError is the sentinel returned by executeMultiAccount when
+// at least one account committed a real purchase while one or more others
+// failed (issue #1014). It is the multi-account analogue of partialPurchaseError:
+// the executor entry points must NOT treat this as a flat failure — the
+// per-account rows already own their authoritative status (partially_completed /
+// completed / failed) and real commitments exist, so an SQS/cron caller must ACK
+// the message (not redeliver) to avoid re-running the fan-out and double-buying
+// the accounts that already succeeded (which #1012's stable key would otherwise
+// dedupe, but the contract should not depend on that second line of defence).
+type multiAccountPartialError struct {
+	committed int
+	errors    []string
+}
+
+func (e *multiAccountPartialError) Error() string {
+	return fmt.Sprintf("multi-account execution: %d account(s) committed, %d failed: %s",
+		e.committed, len(e.errors), strings.Join(e.errors, "; "))
+}
+
+// errAllAccountsFailed is returned by executeMultiAccount when no account
+// committed any purchase — a flat failure that callers may redeliver/retry.
+var errAllAccountsFailed = errors.New("multi-account execution: all accounts failed")
+
 // executeMultiAccount fans out executePurchase across all plan accounts in parallel.
 // Each account gets its own PurchaseExecution record tagged with cloud_account_id.
+//
+// Returns nil when every account fully succeeded; a *multiAccountPartialError
+// when at least one account committed a purchase and at least one failed (so
+// callers ACK rather than redeliver, issue #1014); and errAllAccountsFailed
+// (wrapping the per-account errors) when no account committed anything.
 func (m *Manager) executeMultiAccount(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, accounts []config.CloudAccount) error {
-	results := execution.RunForAccountsWithConcurrency(ctx, accounts, func(ctx context.Context, account config.CloudAccount) (struct{}, error) {
-		return struct{}{}, m.executeForAccount(ctx, baseExec, plan, account)
+	results := execution.RunForAccountsWithConcurrency(ctx, accounts, func(ctx context.Context, account config.CloudAccount) (bool, error) {
+		return m.executeForAccount(ctx, baseExec, plan, account)
 	}, getMaxAccountParallelism())
 
+	committed := 0
 	var errs []string
 	for _, r := range results {
+		if r.Value {
+			// At least one rec committed for this account (full or partial).
+			committed++
+		}
 		if r.Err != nil {
 			errs = append(errs, fmt.Sprintf("account %s: %v", r.AccountID, r.Err))
 		}
 	}
-	if len(errs) > 0 {
-		return fmt.Errorf("multi-account execution: %s", strings.Join(errs, "; "))
+
+	if len(errs) == 0 {
+		return nil
 	}
-	return nil
+	if committed > 0 {
+		// Some accounts committed real purchases; the per-account rows already
+		// recorded the truth. Surface a partial sentinel so the entry points
+		// ack and never mislabel the run as a flat failure (issue #1014).
+		return &multiAccountPartialError{committed: committed, errors: errs}
+	}
+	return fmt.Errorf("%w: %s", errAllAccountsFailed, strings.Join(errs, "; "))
 }
 
 // executeForAccount runs a single plan execution for one cloud account.
 // It creates a new PurchaseExecution record tagged with cloud_account_id, resolves
 // per-account credentials, executes purchases, and saves the result.
-func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, account config.CloudAccount) error {
+//
+// Returns committed=true when at least one rec committed a real purchase for
+// this account (full or partial success), so executeMultiAccount can tell a
+// partial run (some accounts bought) apart from a flat failure and pick the
+// right sentinel (issue #1014). The returned error is non-nil whenever any rec
+// failed, but the authoritative per-account row has already been saved with the
+// correct status (partially_completed / failed) before it surfaces.
+func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.PurchaseExecution, plan *config.PurchasePlan, account config.CloudAccount) (committed bool, err error) {
 	// Create a per-account copy of the execution record with an independent
 	// Recommendations slice so concurrent goroutines don't race on writes.
 	acctID := account.ID
 	acctExec := *baseExec
+	// ExecutionID is the per-account row identity (still a fresh UUID), but the
+	// per-rec provider idempotency token must NOT derive from it: a re-drive of
+	// the root execution would mint new UUIDs and lose the provider dedupe,
+	// double-buying every account (issue #1012 / H1). Seed a STABLE per-account
+	// idempotency key from the root's lineage key + the account ID so a re-drive
+	// reproduces the identical token per account. The account ID is the cloud
+	// account's stable config ID, not the (also stable) ExternalID — either is
+	// durable, but account.ID is the value that uniquely keys the fan-out unit.
 	acctExec.ExecutionID = uuid.New().String()
+	acctExec.IdempotencyKey = idempotencyLineageKey(baseExec) + ":" + account.ID
 	acctExec.CloudAccountID = &acctID
 	recs := make([]config.RecommendationRecord, len(baseExec.Recommendations))
 	copy(recs, baseExec.Recommendations)
@@ -165,7 +221,7 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha
 		acctExec.Status = "failed"
 		acctExec.Error = err.Error()
 		_ = m.config.SavePurchaseExecution(ctx, &acctExec)
-		return fmt.Errorf("credential resolution failed for account %s: %w", account.ID, err)
+		return false, fmt.Errorf("credential resolution failed for account %s: %w", account.ID, err)
 	}
 
 	accountID := account.ExternalID
@@ -196,8 +252,12 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha
 		acctExec.CompletedAt = &now
 	}
 
-	if err := m.config.SavePurchaseExecution(ctx, &acctExec); err != nil {
-		return fmt.Errorf("AUDIT LOSS: failed to save execution record for account %s: %w", account.ID, err)
+	// committed is the gate executeMultiAccount uses to distinguish a partial
+	// run from a flat failure (issue #1014): true when any rec purchased.
+	committed = anyRecPurchased(acctExec.Recommendations)
+
+	if saveErr := m.config.SavePurchaseExecution(ctx, &acctExec); saveErr != nil {
+		return committed, fmt.Errorf("AUDIT LOSS: failed to save execution record for account %s: %w", account.ID, saveErr)
 	}
 
 	// Send the confirmation whenever at least one rec committed (full or
@@ -214,9 +274,9 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha
 	// authoritative per-account row was already saved as partially_completed
 	// (not failed) and the confirmation already went out.
 	if len(purchaseErrors) > 0 {
-		return fmt.Errorf("some purchases failed: %v", purchaseErrors)
+		return committed, fmt.Errorf("some purchases failed: %v", purchaseErrors)
 	}
-	return nil
+	return committed, nil
 }
 
 // resolveSingleAccountProvider derives per-account credentials for the
@@ -458,13 +518,17 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf
 			rec := exec.Recommendations[i]
 			logging.Infof("Purchasing: %dx %s in %s (%s/%s)", rec.Count, rec.ResourceType, rec.Region, rec.Provider, rec.Service)
 			// Derive a deterministic per-rec idempotency token from the
-			// execution ID and this rec's index so a re-drive of a stranded
-			// execution (issue #636) reuses the identical token and the
-			// commitment is never created twice. opts is a value, so this
-			// per-rec copy is safe under the parallel fan-out (no shared
+			// execution's STABLE lineage key (not its mutable ExecutionID)
+			// and this rec's index so a re-drive of a stranded execution
+			// reuses the identical token and the commitment is never created
+			// twice. The lineage key survives Retry and multi-account fan-out
+			// (issue #1012), where the ExecutionID is regenerated; deriving
+			// from the ExecutionID directly defeated the provider guard on
+			// exactly those re-drive paths (#636/#639). opts is a value, so
+			// this per-rec copy is safe under the parallel fan-out (no shared
 			// mutation across goroutines).
 			recOpts := opts
-			recOpts.IdempotencyToken = common.DeriveIdempotencyToken(exec.ExecutionID, i)
+			recOpts.IdempotencyToken = common.DeriveIdempotencyToken(idempotencyLineageKey(exec), i)
 			purchaseResult, err := m.executeSinglePurchase(ctx, rec, provCfg, recOpts)
 			return recPurchaseOutcome{index: i, purchase: purchaseResult, err: err}, nil
 		}, getMaxAccountParallelism())
@@ -542,6 +606,24 @@ func recordHistoryAuditGap(exec *config.PurchaseExecution, commitmentID string,
 	exec.Error = appendErrNote(exec.Error, note)
 }
 
+// idempotencyLineageKey returns the stable anchor the per-rec provider
+// idempotency token is derived from (issue #1012). It prefers the durable
+// IdempotencyKey column, which is generated once at first creation and copied
+// verbatim onto every Retry successor and combined with the account ID for each
+// multi-account fan-out row — so a strand-and-re-drive reproduces the same
+// token and the provider dedupes the purchase. It falls back to ExecutionID
+// only for legacy rows persisted before migration 000066 (IdempotencyKey == "");
+// for a single un-retried execution that fallback is identical to the pre-fix
+// behaviour, and such legacy rows never gain a retry successor that could
+// diverge (the retry handler seeds the successor's key from the predecessor's
+// ExecutionID in that case, preserving the match).
+func idempotencyLineageKey(exec *config.PurchaseExecution) string {
+	if exec.IdempotencyKey != "" {
+		return exec.IdempotencyKey
+	}
+	return exec.ExecutionID
+}
+
 // appendErrNote joins note onto an existing error string with "; ",
 // returning note alone when existing is empty. Shared by the audit-gap and
 // partial-failure paths so a successful-rec audit gap (#621) and a per-rec
diff --git a/internal/purchase/execution_test.go b/internal/purchase/execution_test.go
index 46d777bb5..620876567 100644
--- a/internal/purchase/execution_test.go
+++ b/internal/purchase/execution_test.go
@@ -1582,8 +1582,9 @@ func TestExecuteForAccount_PartialSuccess(t *testing.T) {
 		dashboardURL:    "https://dashboard.example.com",
 	}
 
-	err := manager.executeForAccount(ctx, baseExec, plan, account)
+	committed, err := manager.executeForAccount(ctx, baseExec, plan, account)
 	require.Error(t, err, "the per-rec failure must surface to the aggregator")
+	assert.True(t, committed, "a partial run committed at least one rec, so committed must be true (issue #1014)")
 
 	require.NotNil(t, savedExec, "per-account record must be saved")
 	assert.Equal(t, "partially_completed", savedExec.Status, "partial success must never be 'failed' (double-spend hazard)")
diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go
index 4e92698e1..d783d299a 100644
--- a/internal/purchase/manager.go
+++ b/internal/purchase/manager.go
@@ -126,15 +126,22 @@ func NewManager(cfg ManagerConfig) *Manager {
 }
 
 // finalizeExecution sets the status and completion time on an execution based on the error.
+//
+// Both partial sentinels (single-account *partialPurchaseError and
+// multi-account *multiAccountPartialError) map the root row to
+// "partially_completed", never "failed": real commitments exist and a re-approve
+// would double-buy them (issues #642 / #1014). errAllAccountsFailed falls
+// through to the "failed" default — nothing committed, so a Retry is safe.
 func (m *Manager) finalizeExecution(exec *config.PurchaseExecution, execErr error) {
 	var partial *partialPurchaseError
+	var multiPartial *multiAccountPartialError
 	switch {
 	case execErr == nil:
 		completedAt := time.Now()
 		exec.Status = "completed"
 		exec.CompletedAt = &completedAt
-	case errors.As(execErr, &partial):
-		// #642: at least one rec committed a real purchase while others
+	case errors.As(execErr, &partial), errors.As(execErr, &multiPartial):
+		// At least one rec / account committed a real purchase while others
 		// failed. Never mark such a row "failed" — the commitments are real
 		// and a re-approve would double-buy them. Record the partial outcome
 		// and stamp CompletedAt so the row reads as terminal (the successful
@@ -152,29 +159,82 @@ func (m *Manager) finalizeExecution(exec *config.PurchaseExecution, execErr erro
 	}
 }
 
+// claimAndExecute is the single atomic claim-then-execute funnel for the
+// non-synchronous executor entry points (SQS execute_purchase and the cron
+// ProcessScheduledPurchases sweep), mirroring the guard the synchronous approve
+// path already gets from ApproveAndExecute's CAS (issue #1013).
+//
+// It atomically transitions the row from an executable state
+// (approved/pending/notified) to "running" via TransitionExecutionStatus and
+// only proceeds when it wins that CAS. A lost CAS (row already claimed by a
+// concurrent worker, an overlapping cron tick, an SQS redelivery, or the row
+// vanishing mid-flight) is benign: the function returns (claimed=false, nil) and
+// the caller should ack/skip without re-running the purchase. A real DB error
+// during the claim returns (false, err).
+//
+// On a won claim it runs executeAndFinalize and returns (true, execErr).
+func (m *Manager) claimAndExecute(ctx context.Context, exec *config.PurchaseExecution) (claimed bool, err error) {
+	updated, claimErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved", "pending", "notified"}, "running")
+	if claimErr != nil {
+		if errors.Is(claimErr, config.ErrNotFound) || errors.Is(claimErr, config.ErrExecutionNotInExpectedStatus) {
+			// Benign CAS race-loss: another worker/redelivery already owns this
+			// row (or it was deleted). Ack/skip without executing.
+			logging.Warnf("Skipping execution %s (CAS claim lost to concurrent worker): %v", exec.ExecutionID, claimErr)
+			return false, nil
+		}
+		return false, fmt.Errorf("failed to claim execution %s for execution: %w", exec.ExecutionID, claimErr)
+	}
+	// Carry the committed DB state (status=running, plus any fields refreshed by
+	// the RETURNING clause) onto the caller's struct so the rest of the run
+	// starts from the claimed row rather than the pre-claim snapshot.
+	*exec = *updated
+	return true, m.executeAndFinalize(ctx, exec)
+}
+
+// isMultiAccountAckable reports whether a multi-account execErr should be ACKed
+// (not redelivered/recounted as a flat failure) because at least one account
+// committed a real purchase (issue #1014). A nil error (full success) is also
+// ackable. errAllAccountsFailed is NOT ackable — nothing committed, so a
+// redelivery is safe and useful.
+func isMultiAccountAckable(execErr error) bool {
+	if execErr == nil {
+		return true
+	}
+	var partial *multiAccountPartialError
+	return errors.As(execErr, &partial)
+}
+
 // executeAndFinalize runs a purchase and handles status updates, record saving, and progress.
+//
+// The root execution row is ALWAYS saved with its finalized status — including
+// the multi-account fan-out case (issue #1014 / H2). The per-account fan-out
+// rows are distinct rows that executeForAccount already saved with their own
+// authoritative status; the root row reflects the AGGREGATE outcome
+// (completed / partially_completed / failed, classified by finalizeExecution
+// from the typed sentinel). The previous `!wasMultiAccount` save-skip left the
+// root in whatever pre-execution status it carried, which, now that
+// claimAndExecute claims the root to "running" first (issue #1013), would strand
+// the root row in "running" until the reaper failed it.
 func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseExecution) error {
-	wasMultiAccount, execErr := m.executePurchase(ctx, exec)
+	_, execErr := m.executePurchase(ctx, exec)
 	m.finalizeExecution(exec, execErr)
 	if execErr != nil {
 		logging.Errorf("Failed to execute purchase %s: %v", exec.ExecutionID, execErr)
 	}
-	if !wasMultiAccount {
-		if err := m.config.SavePurchaseExecution(ctx, exec); err != nil {
-			logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err)
-			// Wrap with ErrAuditLoss regardless of whether executePurchase itself
-			// failed. When execErr != nil (provider/partial error), finalizeExecution
-			// stamped a terminal status on the in-memory exec struct, but if
-			// SavePurchaseExecution then failed the DB row is still in "running" --
-			// exactly the stranded-row scenario ErrAuditLoss signals. Preserve the
-			// original execErr as the innermost %w so errors.As/errors.Is can still
-			// reach it from callers (e.g. claimAndRedrive checking ErrAuditLoss).
-			if execErr != nil {
-				execErr = fmt.Errorf("%w: terminal save failed (%v); original execution error: %w",
-					config.ErrAuditLoss, err, execErr)
-			} else {
-				execErr = fmt.Errorf("%w: %w", config.ErrAuditLoss, err)
-			}
+	if err := m.config.SavePurchaseExecution(ctx, exec); err != nil {
+		logging.Errorf("AUDIT LOSS: failed to save execution status: %v", err)
+		// Wrap with ErrAuditLoss regardless of whether executePurchase itself
+		// failed. When execErr != nil (provider/partial error), finalizeExecution
+		// stamped a terminal status on the in-memory exec struct, but if
+		// SavePurchaseExecution then failed the DB row is still in "running" --
+		// exactly the stranded-row scenario ErrAuditLoss signals. Preserve the
+		// original execErr as the innermost %w so errors.As/errors.Is can still
+		// reach it from callers (e.g. claimAndRedrive checking ErrAuditLoss).
+		if execErr != nil {
+			execErr = fmt.Errorf("%w: terminal save failed (%v); original execution error: %w",
+				config.ErrAuditLoss, err, execErr)
+		} else {
+			execErr = fmt.Errorf("%w: %w", config.ErrAuditLoss, err)
 		}
 	}
 	if execErr == nil {
@@ -188,7 +248,9 @@ func (m *Manager) executeAndFinalize(ctx context.Context, exec *config.PurchaseE
 // allRecsSafeToRedrive reports whether every recommendation in the execution
 // can be safely re-driven without risking a double-purchase. A re-drive is safe
 // when the underlying provider purchase API is idempotent under the
-// DeriveIdempotencyToken(exec.ExecutionID, i) scheme used by execution.go.
+// DeriveIdempotencyToken(idempotencyLineageKey(exec), i) scheme used by
+// execution.go. An in-place re-drive (this path) keeps the same row, so the
+// lineage key is unchanged and the token is reproduced exactly.
 //
 // Safe providers / services (issue #639):
 //   - AWS (all services): tag-guard or ClientToken deduplication (#636/#638).
@@ -360,9 +422,12 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution)
 //
 // Idempotent re-drive path (issue #639): all AWS, Azure reservations, and GCP
 // compute service clients derive or look up a deterministic idempotency key
-// from DeriveIdempotencyToken(exec.ExecutionID, i). Re-driving with the same
-// ExecutionID produces the same token, so the cloud provider dedupes the second
-// call and no double-purchase occurs. The row transitions directly to "completed"
+// from DeriveIdempotencyToken(idempotencyLineageKey(exec), i). This is an
+// IN-PLACE re-drive — the same row, so the lineage key is unchanged — which
+// reproduces the same token, so the cloud provider dedupes the second call and
+// no double-purchase occurs. (The retry/fan-out double-buy hole #1012 closes is
+// about NEW rows minting a fresh ExecutionID; that does not apply here.) The row
+// transitions directly to "completed"
 // (or "failed"/"partially_completed" on a genuine error), bypassing the manual
 // Retry step required by the old safe-fail path. See allRecsSafeToRedrive for
 // which provider/service combinations are eligible.
@@ -370,6 +435,7 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution)
 // Safe-fail path: Azure savings-plans recs are excluded because the OrderAlias
 // API uses a timestamp-based alias name with no idempotency key. Executions
 // without a stable ExecutionID (legacy rows) also fall through because
+// idempotencyLineageKey(exec) falls back to "" for them and
 // DeriveIdempotencyToken("", i) would produce the same token set for every
 // such row. These fall through to the original behaviour: the row is atomically
 // transitioned to "failed" so it surfaces in History and can be Retry-ed by
@@ -390,10 +456,11 @@ func (m *Manager) RecoverStrandedApprovals(ctx context.Context) (int, error) {
 		exec := &stranded[i]
 
 		// Idempotent re-drive path (issue #639): all recs honour
-		// opts.IdempotencyToken via DeriveIdempotencyToken(exec.ExecutionID, i),
-		// so a second call with the same ExecutionID is a safe no-op on the
-		// provider side. The ExecutionID must be non-empty to derive a unique
-		// token; an empty ID would map every legacy row to the same token set.
+		// opts.IdempotencyToken via DeriveIdempotencyToken(idempotencyLineageKey(exec), i),
+		// so a second in-place call on the same row is a safe no-op on the
+		// provider side. The ExecutionID must be non-empty so the lineage key
+		// (or its ExecutionID fallback for legacy rows) is unique; an empty ID
+		// would map every legacy row to the same token set.
 		if allRecsSafeToRedrive(exec) && exec.ExecutionID != "" {
 			counted, driveErr := m.claimAndRedrive(ctx, exec)
 			if driveErr != nil {
@@ -441,17 +508,22 @@ func (m *Manager) ProcessScheduledPurchases(ctx context.Context) (*ProcessResult
 	processed := 0
 	executed := 0
 	failed := 0
-	var errors []string
+	var errs []string
 
-	for _, exec := range executions {
+	for i := range executions {
+		exec := executions[i]
 		// Check if it's time to execute
 		if exec.ScheduledDate.After(now) {
 			logging.Debugf("Execution %s not yet due (scheduled for %s)", exec.ExecutionID, exec.ScheduledDate)
 			continue
 		}
 
-		// Skip if cancelled or already completed
-		if exec.Status == "cancelled" || exec.Status == "completed" {
+		// Positive allowlist (issue #1013 / M2): only the two executable
+		// pre-purchase states proceed. The query already filters to
+		// pending/notified, but a row another worker transitioned in the gap
+		// between SELECT and here (approved/running/failed/...) must be skipped.
+		// The atomic claim below is the real guard; this is defence-in-depth.
+		if exec.Status != "pending" && exec.Status != "notified" {
 			continue
 		}
 
@@ -459,13 +531,33 @@ func (m *Manager) ProcessScheduledPurchases(ctx context.Context) (*ProcessResult
 
 		logging.Infof("Executing scheduled purchase: %s", exec.ExecutionID)
 
-		// Execute the purchase and handle post-execution bookkeeping.
-		if execErr := m.executeAndFinalize(ctx, &exec); execErr != nil {
-			failed++
-			errors = append(errors, fmt.Sprintf("%s: %v", exec.ExecutionID, execErr))
-		} else {
+		// Atomically claim the row before executing (issue #1013). Overlapping
+		// cron ticks (a tick that runs longer than the interval, EventBridge
+		// duplicate/overlapping deliveries, or cron racing the SQS path) would
+		// otherwise both execute the same due row. claimAndExecute CASes the row
+		// to "running" and only the winner runs; a lost claim is skipped without
+		// re-executing.
+		claimed, execErr := m.claimAndExecute(ctx, &exec)
+		if !claimed {
+			// execErr != nil here is a real DB error during the claim (count as
+			// failed); execErr == nil is a benign CAS race-loss (skip silently).
+			if execErr != nil {
+				failed++
+				errs = append(errs, fmt.Sprintf("%s: claim failed: %v", exec.ExecutionID, execErr))
+			}
+			continue
+		}
+
+		// A multi-account run where at least one account committed is a success
+		// for ack purposes (issue #1014): the per-account rows own the truth and
+		// re-running would double-buy. Only a genuine failure (nothing
+		// committed) is counted/surfaced.
+		if isMultiAccountAckable(execErr) {
 			executed++
+			continue
 		}
+		failed++
+		errs = append(errs, fmt.Sprintf("%s: %v", exec.ExecutionID, execErr))
 	}
 
 	return &ProcessResult{
@@ -473,6 +565,6 @@ func (m *Manager) ProcessScheduledPurchases(ctx context.Context) (*ProcessResult
 		Executed:  executed,
 		Failed:    failed,
 		Recovered: recovered,
-		Errors:    errors,
+		Errors:    errs,
 	}, nil
 }
diff --git a/internal/purchase/manager_test.go b/internal/purchase/manager_test.go
index 2afe531bb..50466f17f 100644
--- a/internal/purchase/manager_test.go
+++ b/internal/purchase/manager_test.go
@@ -205,8 +205,13 @@ func TestManager_ProcessScheduledPurchases_DuePurchase(t *testing.T) {
 		},
 	}
 
+	// claimAndExecute CASes the due row to "running" before executing (issue #1013).
+	claimedExec := executions[0]
+	claimedExec.Status = "running"
 	mockStore.On("GetStaleApprovedExecutions", ctx, mock.Anything).Return([]config.PurchaseExecution{}, nil)
 	mockStore.On("GetPendingExecutions", ctx).Return(executions, nil)
+	mockStore.On("TransitionExecutionStatus", ctx, "exec-123",
+		[]string{"approved", "pending", "notified"}, "running").Return(&claimedExec, nil)
 	mockStore.On("GetPurchasePlan", ctx, "plan-456").Return(plan, nil).Twice()
 	mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
 	mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
@@ -297,8 +302,12 @@ func TestManager_ProcessScheduledPurchases_ExecutionFails(t *testing.T) {
 		},
 	}
 
+	claimedExec := executions[0]
+	claimedExec.Status = "running"
 	mockStore.On("GetStaleApprovedExecutions", ctx, mock.Anything).Return([]config.PurchaseExecution{}, nil)
 	mockStore.On("GetPendingExecutions", ctx).Return(executions, nil)
+	mockStore.On("TransitionExecutionStatus", ctx, "exec-123",
+		[]string{"approved", "pending", "notified"}, "running").Return(&claimedExec, nil)
 	mockStore.On("GetPurchasePlan", ctx, "plan-456").Return(nil, errors.New("plan not found")).Once()
 	mockStore.On("SavePurchaseExecution", ctx, mock.AnythingOfType("*config.PurchaseExecution")).Return(nil)
 	// updatePlanProgress is NOT called when execution fails
diff --git a/internal/purchase/messages.go b/internal/purchase/messages.go
index 3da2ed18a..ea16e4ec2 100644
--- a/internal/purchase/messages.go
+++ b/internal/purchase/messages.go
@@ -91,28 +91,36 @@ func (m *Manager) handleExecutePurchase(ctx context.Context, msg AsyncMessage) e
 		return fmt.Errorf("execution not found: %s", msg.ExecutionID)
 	}
 
-	// Only execute if approved or pending (auto-approved)
-	if execution.Status != "approved" && execution.Status != "pending" {
-		logging.Warnf("Execution %s not in executable state (status: %s), skipping", msg.ExecutionID, execution.Status)
-		return nil
-	}
-
 	logging.Infof("Executing purchase from async message: %s", msg.ExecutionID)
-	wasMultiAccount, purchaseErr := m.executePurchase(ctx, execution)
-	m.finalizeExecution(execution, purchaseErr)
 
-	if !wasMultiAccount {
-		if saveErr := m.config.SavePurchaseExecution(ctx, execution); saveErr != nil {
-			logging.Errorf("Failed to save execution status: %v", saveErr)
-			return fmt.Errorf("failed to save execution status for %s: %w", msg.ExecutionID, saveErr)
+	// Atomically claim the row before touching the cloud (issue #1013). SQS
+	// delivery is at-least-once: a redelivered execute_purchase message (slow
+	// purchase whose visibility timeout expired, partial-batch replay, operator
+	// re-drive) would otherwise pass the old non-atomic status check and execute
+	// the same row a second time concurrently. claimAndExecute CASes
+	// approved/pending/notified -> running and only proceeds on a won claim; a
+	// lost claim is a benign duplicate that we ack (return nil) without
+	// re-executing.
+	claimed, execErr := m.claimAndExecute(ctx, execution)
+	if !claimed {
+		// Either a benign CAS race-loss (execErr == nil — ack the duplicate) or
+		// a real DB error during the claim (execErr != nil — surface it so the
+		// message is redelivered, since nothing was executed).
+		if execErr != nil {
+			return fmt.Errorf("failed to claim execution %s: %w", msg.ExecutionID, execErr)
 		}
+		return nil
 	}
-	if purchaseErr == nil {
-		if err := m.updatePlanProgress(ctx, execution.PlanID); err != nil {
-			logging.Errorf("Failed to update plan progress: %v", err)
-		}
+
+	// A multi-account run where at least one account committed must be ACKed,
+	// not redelivered (issue #1014): redelivery would re-run the fan-out and,
+	// absent the per-account idempotency key (#1012), double-buy the accounts
+	// that already succeeded. errAllAccountsFailed (nothing committed) and
+	// single-account errors fall through and are returned so SQS redelivers.
+	if isMultiAccountAckable(execErr) {
+		return nil
 	}
-	return purchaseErr
+	return execErr
 }
 
 // handleApproveMessage processes an approve message.
diff --git a/internal/purchase/messages_test.go b/internal/purchase/messages_test.go
index 3af452b08..6cae5670a 100644
--- a/internal/purchase/messages_test.go
+++ b/internal/purchase/messages_test.go
@@ -2,6 +2,7 @@ package purchase
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/LeanerCloud/CUDly/internal/config"
@@ -118,6 +119,13 @@ func TestManager_ProcessMessage(t *testing.T) {
 			Status:      "cancelled",
 		}
 		mockStore.On("GetExecutionByID", ctx, "exec-123").Return(execution, nil)
+		// The claim CAS rejects a non-executable status (issue #1013): the row
+		// is "cancelled", not in [approved,pending,notified], so the CAS loses
+		// and returns ErrExecutionNotInExpectedStatus — a benign skip that the
+		// handler acks without error.
+		mockStore.On("TransitionExecutionStatus", ctx, "exec-123",
+			[]string{"approved", "pending", "notified"}, "running").
+			Return(nil, fmt.Errorf("%w: cancelled", config.ErrExecutionNotInExpectedStatus))
 
 		err := manager.ProcessMessage(ctx, `{"type": "execute_purchase", "execution_id": "exec-123"}`)
 		// Should skip without error when status is not executable
diff --git a/internal/purchase/money_path_regression_test.go b/internal/purchase/money_path_regression_test.go
new file mode 100644
index 000000000..047db4b28
--- /dev/null
+++ b/internal/purchase/money_path_regression_test.go
@@ -0,0 +1,358 @@
+package purchase
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+
+	"github.com/LeanerCloud/CUDly/internal/config"
+	"github.com/LeanerCloud/CUDly/pkg/common"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+)
+
+// awsAccessKeyCredStore returns a credential store that resolves a static AWS
+// access-key blob, so the per-account credential resolution in the fan-out path
+// succeeds without real cloud calls.
+func awsAccessKeyCredStore() *MockCredentialStore {
+	return &MockCredentialStore{
+		LoadRawFn: func(_ context.Context, _, _ string) ([]byte, error) {
+			return []byte(`{"access_key_id":"AKIAIOSFODNN7EXAMPLE","secret_access_key":"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}`), nil
+		},
+	}
+}
+
+// captureIdempotencyTokens runs a single-account execution for exec and returns
+// the IdempotencyToken each PurchaseCommitment call received, keyed by resource
+// type. It wires a provider that records opts.IdempotencyToken on every call.
+func captureIdempotencyTokens(t *testing.T, exec *config.PurchaseExecution) map[string]string {
+	t.Helper()
+	ctx := context.Background()
+
+	mockStore := new(MockConfigStore)
+	mockEmail := new(MockEmailSender)
+	mockFactory := new(MockProviderFactory)
+	mockProviderInst := new(MockProvider)
+	mockServiceClient := new(MockServiceClient)
+
+	plan := &config.PurchasePlan{Name: "Direct purchase"}
+
+	mockStore.SavePurchaseExecutionFn = func(_ context.Context, _ *config.PurchaseExecution) error { return nil }
+	mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
+	mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
+
+	mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil)
+	mockProviderInst.On("GetServiceClient", mock.Anything, common.ServiceEC2, mock.Anything).Return(mockServiceClient, nil)
+
+	var mu sync.Mutex
+	tokens := map[string]string{}
+	mockServiceClient.On("PurchaseCommitment", mock.Anything,
+		mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions"),
+	).Run(func(args mock.Arguments) {
+		rec := args.Get(1).(common.Recommendation)
+		opts := args.Get(2).(common.PurchaseOptions)
+		mu.Lock()
+		tokens[rec.ResourceType] = opts.IdempotencyToken
+		mu.Unlock()
+	}).Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil)
+
+	manager := &Manager{
+		config:          mockStore,
+		email:           mockEmail,
+		providerFactory: mockFactory,
+		dashboardURL:    "https://dashboard.example.com",
+	}
+
+	// provCfg is nil: the mock factory matches mock.Anything for it and the
+	// real factory is never reached.
+	_, _, errs := manager.processPurchaseRecommendations(ctx, exec, plan, "111111111111", nil)
+	require.Empty(t, errs, "all recs must commit so the captured token reflects a real purchase")
+	return tokens
+}
+
+// TestRetryReusesIdempotencyToken is the issue #1012 regression guard: a retry
+// of a failed-but-landed execution must derive the SAME per-rec provider
+// idempotency token as the original attempt, so the provider dedupes and the
+// commitment is never bought twice.
+//
+// Pre-fix, the token derived purely from the execution_id, which the retry
+// regenerates as a fresh UUID — so this test would observe DIFFERENT tokens and
+// fail. Post-fix, the retry copies the stable IdempotencyKey verbatim, so the
+// tokens match.
+func TestRetryReusesIdempotencyToken(t *testing.T) {
+	recs := []config.RecommendationRecord{
+		{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true},
+	}
+
+	original := &config.PurchaseExecution{
+		ExecutionID:     "exec-original",
+		IdempotencyKey:  "stable-lineage-key-abc",
+		Source:          common.PurchaseSourceWeb,
+		Recommendations: append([]config.RecommendationRecord(nil), recs...),
+	}
+	// The retry successor: a FRESH ExecutionID (as persistRetryExecution mints),
+	// but the SAME IdempotencyKey copied verbatim from the predecessor.
+	retry := &config.PurchaseExecution{
+		ExecutionID:     "exec-retry-NEW-uuid",
+		IdempotencyKey:  "stable-lineage-key-abc",
+		Source:          common.PurchaseSourceWeb,
+		Recommendations: append([]config.RecommendationRecord(nil), recs...),
+	}
+
+	origTokens := captureIdempotencyTokens(t, original)
+	retryTokens := captureIdempotencyTokens(t, retry)
+
+	require.NotEmpty(t, origTokens["m5.large"], "original must derive a non-empty token")
+	assert.Equal(t, origTokens["m5.large"], retryTokens["m5.large"],
+		"retry must derive the SAME idempotency token as the original (issue #1012) despite a fresh ExecutionID")
+	// Sanity: the token is derived from the stable lineage key, not the exec ID.
+	assert.Equal(t, common.DeriveIdempotencyToken("stable-lineage-key-abc", 0), origTokens["m5.large"])
+}
+
+// TestLegacyRowFallsBackToExecutionID guards the migration-000066 legacy path:
+// a row with no IdempotencyKey (NULL column) must derive its token from the
+// ExecutionID, identical to the pre-fix behaviour for a single un-retried
+// execution, so old in-flight rows keep working.
+func TestLegacyRowFallsBackToExecutionID(t *testing.T) {
+	legacy := &config.PurchaseExecution{
+		ExecutionID: "legacy-exec-id",
+		// IdempotencyKey deliberately empty (legacy NULL).
+		Source: common.PurchaseSourceWeb,
+		Recommendations: []config.RecommendationRecord{
+			{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true},
+		},
+	}
+	tokens := captureIdempotencyTokens(t, legacy)
+	assert.Equal(t, common.DeriveIdempotencyToken("legacy-exec-id", 0), tokens["m5.large"],
+		"a legacy row (no idempotency_key) must fall back to ExecutionID for the token")
+}
+
+// TestMultiAccountSeedsStablePerAccountKey is the issue #1012 / H1 guard for the
+// multi-account fan-out: a re-drive of a multi-account plan must reproduce the
+// same per-account idempotency key (root lineage key + account ID), NOT a fresh
+// UUID. Pre-fix, each account row minted a random UUID at execution time, so a
+// second run derived a different token and double-bought. This test asserts the
+// per-account IdempotencyKey is deterministic across two runs of the same root.
+func TestMultiAccountSeedsStablePerAccountKey(t *testing.T) {
+	ctx := context.Background()
+	accounts := []config.CloudAccount{
+		{ID: "acct-A", Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys"},
+		{ID: "acct-B", Name: "B", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "access_keys"},
+	}
+
+	runOnce := func() map[string]string {
+		mockStore := new(MockConfigStore)
+		mockEmail := new(MockEmailSender)
+		mockFactory := new(MockProviderFactory)
+		mockProviderInst := new(MockProvider)
+		mockServiceClient := new(MockServiceClient)
+
+		baseExec := &config.PurchaseExecution{
+			ExecutionID:    "root-exec",
+			IdempotencyKey: "root-lineage",
+			PlanID:         "plan-x",
+			Source:         common.PurchaseSourceWeb,
+			Recommendations: []config.RecommendationRecord{
+				{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true},
+			},
+		}
+		plan := &config.PurchasePlan{ID: "plan-x", Name: "Plan X"}
+
+		var mu sync.Mutex
+		perAccountKey := map[string]string{}
+		mockStore.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error {
+			mu.Lock()
+			if e.CloudAccountID != nil {
+				perAccountKey[*e.CloudAccountID] = e.IdempotencyKey
+			}
+			mu.Unlock()
+			return nil
+		}
+		mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
+		mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
+
+		mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil)
+		mockProviderInst.On("GetServiceClient", mock.Anything, common.ServiceEC2, mock.Anything).Return(mockServiceClient, nil)
+		mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything).
+			Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil)
+
+		manager := &Manager{
+			config:          mockStore,
+			email:           mockEmail,
+			providerFactory: mockFactory,
+			credStore:       awsAccessKeyCredStore(),
+			dashboardURL:    "https://dashboard.example.com",
+		}
+		require.NoError(t, manager.executeMultiAccount(ctx, baseExec, plan, accounts))
+		return perAccountKey
+	}
+
+	run1 := runOnce()
+	run2 := runOnce()
+
+	require.Equal(t, "root-lineage:acct-A", run1["acct-A"], "per-account key must be lineage+accountID, not a UUID")
+	require.Equal(t, "root-lineage:acct-B", run1["acct-B"])
+	assert.Equal(t, run1, run2, "two runs of the same root must reproduce identical per-account idempotency keys (issue #1012/H1)")
+}
+
+// TestSQSRedeliveryDoesNotDoubleExecute is the issue #1013 (C2) regression
+// guard: a redelivered execute_purchase SQS message must NOT execute the row a
+// second time. The first delivery wins the CAS to "running"; the redelivery
+// loses the CAS (the row is no longer in an executable state) and is acked
+// without touching the cloud.
+//
+// Pre-fix, handleExecutePurchase only checked status in (approved,pending) with
+// no atomic transition, so both deliveries would execute. This test asserts the
+// purchase runs exactly once and the second delivery is a benign no-op.
+func TestSQSRedeliveryDoesNotDoubleExecute(t *testing.T) {
+	ctx := context.Background()
+	mockStore := new(MockConfigStore)
+	mockEmail := new(MockEmailSender)
+	mockFactory := new(MockProviderFactory)
+	mockProviderInst := new(MockProvider)
+	mockServiceClient := new(MockServiceClient)
+
+	newPending := func() *config.PurchaseExecution {
+		return &config.PurchaseExecution{
+			ExecutionID:    "exec-dup",
+			IdempotencyKey: "lineage-dup",
+			Status:         "pending",
+			Recommendations: []config.RecommendationRecord{
+				{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true},
+			},
+		}
+	}
+
+	// Both deliveries read the row as "pending" from the DB (at-least-once SQS:
+	// pre-fix nothing CASes it to running before the cloud call, so a redelivery
+	// re-reads a still-claimable row). Each GetExecutionByID returns a FRESH
+	// pending copy, faithfully modelling the real double-delivery scenario.
+	mockStore.On("GetExecutionByID", ctx, "exec-dup").Return(newPending(), nil).Once()
+	mockStore.On("GetExecutionByID", ctx, "exec-dup").Return(newPending(), nil).Once()
+
+	// First claim wins: pending -> running, returns the running row. The second
+	// claim (redelivery) loses with ErrExecutionNotInExpectedStatus, because the
+	// row already left the executable set. testify replays Once() expectations in
+	// declaration order, so the first call gets the win and the second the loss.
+	running := newPending()
+	running.Status = "running"
+	mockStore.On("TransitionExecutionStatus", ctx, "exec-dup",
+		[]string{"approved", "pending", "notified"}, "running").Return(running, nil).Once()
+	mockStore.On("TransitionExecutionStatus", ctx, "exec-dup",
+		[]string{"approved", "pending", "notified"}, "running").
+		Return(nil, fmt.Errorf("%w: row already running", config.ErrExecutionNotInExpectedStatus)).Once()
+
+	mockStore.SavePurchaseExecutionFn = func(_ context.Context, _ *config.PurchaseExecution) error { return nil }
+	mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
+	mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
+	mockStore.On("GetPurchasePlan", ctx, mock.Anything).Return(&config.PurchasePlan{Name: "p"}, nil).Maybe()
+
+	mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil)
+	mockProviderInst.On("GetServiceClient", mock.Anything, common.ServiceEC2, mock.Anything).Return(mockServiceClient, nil)
+
+	var purchaseCalls int
+	var mu sync.Mutex
+	mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything).
+		Run(func(mock.Arguments) { mu.Lock(); purchaseCalls++; mu.Unlock() }).
+		Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil)
+
+	manager := &Manager{
+		config:          mockStore,
+		email:           mockEmail,
+		providerFactory: mockFactory,
+		dashboardURL:    "https://dashboard.example.com",
+	}
+
+	msg := AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: "exec-dup"}
+	require.NoError(t, manager.handleExecutePurchase(ctx, msg), "first delivery executes cleanly")
+	require.NoError(t, manager.handleExecutePurchase(ctx, msg), "redelivery is a benign no-op (acked)")
+
+	mu.Lock()
+	defer mu.Unlock()
+	assert.Equal(t, 1, purchaseCalls, "the cloud purchase must run EXACTLY once across both deliveries (issue #1013)")
+}
+
+// TestMultiAccountPartialSuccessIsAcked is the issue #1014 regression guard: a
+// multi-account run where at least one account committed must NOT surface as a
+// flat failure to the SQS handler (which would trigger redelivery and a
+// double-buy). The handler must return nil (ack) when committed >= 1.
+func TestMultiAccountPartialSuccessIsAcked(t *testing.T) {
+	ctx := context.Background()
+	mockStore := new(MockConfigStore)
+	mockEmail := new(MockEmailSender)
+	mockFactory := new(MockProviderFactory)
+	mockProviderInst := new(MockProvider)
+	mockServiceClient := new(MockServiceClient)
+
+	accounts := []config.CloudAccount{
+		{ID: "acct-ok", Name: "OK", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys"},
+		// acct-bad uses a non-access_keys auth mode with no STS client wired, so
+		// its credential resolution fails deterministically (committed=false) —
+		// modelling "one account in the fan-out fails" without racing the mock.
+		{ID: "acct-bad", Name: "BAD", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "role_arn"},
+	}
+
+	exec := &config.PurchaseExecution{
+		ExecutionID:    "root-partial",
+		IdempotencyKey: "lineage-partial",
+		Status:         "pending",
+		PlanID:         "plan-x",
+		Recommendations: []config.RecommendationRecord{
+			{Provider: "aws", Service: "ec2", ResourceType: "m5.large", Region: "us-east-1", Count: 1, UpfrontCost: 300, Selected: true},
+		},
+	}
+	plan := &config.PurchasePlan{ID: "plan-x", Name: "Plan X"}
+
+	mockStore.On("GetExecutionByID", ctx, "root-partial").Return(exec, nil)
+	running := *exec
+	running.Status = "running"
+	mockStore.On("TransitionExecutionStatus", ctx, "root-partial",
+		[]string{"approved", "pending", "notified"}, "running").Return(&running, nil)
+	mockStore.On("GetPurchasePlan", ctx, "plan-x").Return(plan, nil)
+	// GetPlanAccounts is served by the Fn hook, not a testify expectation.
+	mockStore.GetPlanAccountsFn = func(_ context.Context, _ string) ([]config.CloudAccount, error) {
+		return accounts, nil
+	}
+	// updatePlanProgress only runs on a fully-clean run (execErr == nil); a
+	// partial multi-account run skips it. Marked Maybe so a (non-deterministic)
+	// all-success ordering of the two account goroutines doesn't fail the mock.
+	mockStore.On("UpdatePurchasePlan", ctx, mock.AnythingOfType("*config.PurchasePlan")).Return(nil).Maybe()
+
+	var savedRoot *config.PurchaseExecution
+	mockStore.SavePurchaseExecutionFn = func(_ context.Context, e *config.PurchaseExecution) error {
+		if e.CloudAccountID == nil {
+			c := *e
+			savedRoot = &c
+		}
+		return nil
+	}
+	mockStore.On("SavePurchaseHistory", ctx, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil)
+	mockEmail.On("SendPurchaseConfirmation", ctx, mock.AnythingOfType("email.NotificationData")).Return(nil)
+
+	mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil)
+	mockProviderInst.On("GetServiceClient", mock.Anything, common.ServiceEC2, mock.Anything).Return(mockServiceClient, nil)
+	// Only acct-ok reaches a purchase (acct-bad fails at credential resolution),
+	// so exactly one PurchaseCommitment call commits — the run is a partial
+	// success: 1 account committed, 1 account failed.
+	mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.Anything).
+		Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil).Once()
+
+	manager := &Manager{
+		config:          mockStore,
+		email:           mockEmail,
+		providerFactory: mockFactory,
+		credStore:       awsAccessKeyCredStore(),
+		dashboardURL:    "https://dashboard.example.com",
+	}
+
+	msg := AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: "root-partial"}
+	err := manager.handleExecutePurchase(ctx, msg)
+	require.NoError(t, err,
+		"a multi-account run with >=1 committed account must be ACKed (return nil), not redelivered (issue #1014)")
+
+	require.NotNil(t, savedRoot, "the root row must be saved with its aggregate status (H2)")
+	assert.Equal(t, "partially_completed", savedRoot.Status,
+		"root must reflect partial success, never 'failed' (double-spend mislabel #642/#1014)")
+}

From 479c4a9fbdbf0a8825207850fe4c2f919733aa00 Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 04:02:05 -0700
Subject: [PATCH 2/7] refactor(config): extract execution nullable-time mapping
 to satisfy gocyclo

Adding idempotency_key handling pushed scanExecutionRows to cyclomatic
complexity 11, one over the budget of 10. Pull the nullable-timestamp
mapping into applyExecutionNullableTimes, a behavior-preserving helper,
bringing scanExecutionRows back under the limit. No logic change.
---
 internal/config/store_postgres.go | 38 ++++++++++++++++++-------------
 1 file changed, 22 insertions(+), 16 deletions(-)

diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go
index d599481cc..c49e548b2 100644
--- a/internal/config/store_postgres.go
+++ b/internal/config/store_postgres.go
@@ -1297,22 +1297,7 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 			return nil, fmt.Errorf("failed to unmarshal recommendations: %w", err)
 		}
 
-		// Handle nullable timestamps
-		if notifSent.Valid {
-			exec.NotificationSent = &notifSent.Time
-		}
-		if completedAt.Valid {
-			exec.CompletedAt = &completedAt.Time
-		}
-		if expiresAt.Valid {
-			exec.TTL = ttlFromTime(expiresAt.Time)
-		}
-		if tokenExpiresAt.Valid {
-			exec.ApprovalTokenExpiresAt = &tokenExpiresAt.Time
-		}
-		if executedAt.Valid {
-			exec.ExecutedAt = &executedAt.Time
-		}
+		applyExecutionNullableTimes(&exec, notifSent, completedAt, expiresAt, tokenExpiresAt, executedAt)
 
 		executions = append(executions, exec)
 	}
@@ -1320,6 +1305,27 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 	return executions, rows.Err()
 }
 
+// applyExecutionNullableTimes maps the nullable timestamp columns onto exec.
+// It pulls the per-field NULL handling out of scanExecutionRows to keep that
+// function under the cyclomatic limit.
+func applyExecutionNullableTimes(exec *PurchaseExecution, notifSent, completedAt, expiresAt, tokenExpiresAt, executedAt sql.NullTime) {
+	if notifSent.Valid {
+		exec.NotificationSent = &notifSent.Time
+	}
+	if completedAt.Valid {
+		exec.CompletedAt = &completedAt.Time
+	}
+	if expiresAt.Valid {
+		exec.TTL = ttlFromTime(expiresAt.Time)
+	}
+	if tokenExpiresAt.Valid {
+		exec.ApprovalTokenExpiresAt = &tokenExpiresAt.Time
+	}
+	if executedAt.Valid {
+		exec.ExecutedAt = &executedAt.Time
+	}
+}
+
 // CleanupOldExecutions deletes purchase executions older than retentionDays.
 //
 // Two independent cleanup branches, each with its own retention window so

From bab0a3bba19da3186f582776a257d13dae963d91 Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 04:02:16 -0700
Subject: [PATCH 3/7] style(exchange): gofmt struct field alignment in
 reshape_crossfamily_test

The pre-commit gofmt hook runs against the PR merge ref and flagged
unaligned struct literal fields in reshape_crossfamily_test.go (carried
over from an earlier merge). Re-run gofmt to unblock CI. Whitespace only.
---
 pkg/exchange/reshape_crossfamily_test.go | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/pkg/exchange/reshape_crossfamily_test.go b/pkg/exchange/reshape_crossfamily_test.go
index 7d4eb7f60..30f6d372d 100644
--- a/pkg/exchange/reshape_crossfamily_test.go
+++ b/pkg/exchange/reshape_crossfamily_test.go
@@ -465,7 +465,7 @@ func TestCompositeScore_SameGenOutranksTermMismatch(t *testing.T) {
 	}
 	// m6i.xlarge: same "m" prefix (gen jump bonus), exact NF, high confidence.
 	nearPerfect := OfferingOption{
-		InstanceType:        "m6i.xlarge",
+		InstanceType:         "m6i.xlarge",
 		EffectiveMonthlyCost: 85, // slightly more expensive
 		NormalizationFactor:  8,
 		SavingsAbs:           floatPtr(220),
@@ -473,7 +473,7 @@ func TestCompositeScore_SameGenOutranksTermMismatch(t *testing.T) {
 	}
 	// r5.xlarge: different prefix (no family gen bonus), same NF, no confidence.
 	crossFamily := OfferingOption{
-		InstanceType:        "r5.xlarge",
+		InstanceType:         "r5.xlarge",
 		EffectiveMonthlyCost: 80, // same cost as source -- better raw price
 		NormalizationFactor:  8,
 	}
@@ -497,13 +497,13 @@ func TestCompositeScore_SameArchOutranksCrossArch(t *testing.T) {
 	}
 	// c6i: same "c" prefix (family gen bonus), Intel x86 (same arch).
 	sameArch := OfferingOption{
-		InstanceType:        "c6i.xlarge",
+		InstanceType:         "c6i.xlarge",
 		EffectiveMonthlyCost: 90,
 		NormalizationFactor:  8,
 	}
 	// c6g: same "c" prefix (family gen bonus), Graviton ARM (cross arch).
 	crossArch := OfferingOption{
-		InstanceType:        "c6g.xlarge",
+		InstanceType:         "c6g.xlarge",
 		EffectiveMonthlyCost: 90,
 		NormalizationFactor:  8,
 	}
@@ -522,14 +522,14 @@ func TestCompositeScore_HighConfidenceOutranksLow(t *testing.T) {
 		MonthlyCost:         70,
 	}
 	highConf := OfferingOption{
-		InstanceType:        "r5.xlarge",
+		InstanceType:         "r5.xlarge",
 		EffectiveMonthlyCost: 70,
 		NormalizationFactor:  8,
 		SavingsAbs:           floatPtr(250),
 		RecommendationCount:  4,
 	}
 	lowConf := OfferingOption{
-		InstanceType:        "r5.xlarge",
+		InstanceType:         "r5.xlarge",
 		EffectiveMonthlyCost: 70,
 		NormalizationFactor:  8,
 		SavingsAbs:           floatPtr(10),
@@ -550,13 +550,13 @@ func TestCompositeScore_AbsentSavingsIsNeutral(t *testing.T) {
 		MonthlyCost:         70,
 	}
 	absentSavings := OfferingOption{
-		InstanceType:        "r6i.xlarge",
+		InstanceType:         "r6i.xlarge",
 		EffectiveMonthlyCost: 70,
 		NormalizationFactor:  8,
 		SavingsAbs:           nil, // not supplied
 	}
 	lowConf := OfferingOption{
-		InstanceType:        "r6i.xlarge",
+		InstanceType:         "r6i.xlarge",
 		EffectiveMonthlyCost: 70,
 		NormalizationFactor:  8,
 		SavingsAbs:           floatPtr(5), // explicitly low confidence

From 1adfa8cb8f523ee4349857f757b85564249f5c31 Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 04:08:35 -0700
Subject: [PATCH 4/7] fix(config): project idempotency_key in
 GetPlannedExecutions SELECT

GetPlannedExecutions was the only execution-reading query whose SELECT
still stopped at approval_token_expires_at (22 columns) after migration
000066 added idempotency_key. Every other query and the shared
scanExecutionRows Scan target expect 26 columns, so the planned-purchase
list path (handler_purchases.go) would fail at runtime with "failed to
scan execution". Add the four trailing columns (executed_by_user_id,
executed_at, pre_approval_skip_reason, idempotency_key) to align the
projection with the scan.

Add TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns, which fails
on the pre-fix projection (the regexp query matcher does not match
idempotency_key) and passes once the columns are present, mirroring the
real scan-failure path. Also assert IdempotencyKey scan outcomes in the
GetExecutionByID NULL and non-NULL tests so a scan-order regression on
that field cannot slip through.
---
 internal/config/store_postgres.go             |  4 +-
 .../config/store_postgres_pgxmock_test.go     | 59 +++++++++++++++++++
 2 files changed, 62 insertions(+), 1 deletion(-)

diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go
index c49e548b2..dbd07a461 100644
--- a/internal/config/store_postgres.go
+++ b/internal/config/store_postgres.go
@@ -995,7 +995,9 @@ func (s *PostgresStore) GetPlannedExecutions(ctx context.Context, statuses []str
 		       total_upfront_cost, estimated_savings, completed_at, error, expires_at,
 		       cloud_account_id, source, approved_by, cancelled_by, capacity_percent,
 		       created_by_user_id, retry_execution_id, retry_attempt_n,
-		       approval_token_expires_at
+		       approval_token_expires_at,
+		       executed_by_user_id, executed_at, pre_approval_skip_reason,
+		       idempotency_key
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		ORDER BY scheduled_date ASC NULLS LAST, id ASC
diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go
index 7fc9802a3..8c6a5e3b3 100644
--- a/internal/config/store_postgres_pgxmock_test.go
+++ b/internal/config/store_postgres_pgxmock_test.go
@@ -461,6 +461,10 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) {
 	// columns will catch.
 	assert.Nil(t, exec.RetryExecutionID)
 	assert.Equal(t, 0, exec.RetryAttemptN)
+	// idempotency_key scan-outcome guard (CR): the NULL path (legacy rows
+	// before migration 000066) must leave IdempotencyKey empty so derivation
+	// falls back to ExecutionID (issue #1012).
+	assert.Equal(t, "", exec.IdempotencyKey)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
@@ -511,6 +515,61 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) {
 	require.NotNil(t, exec.RetryExecutionID)
 	assert.Equal(t, "exec-3", *exec.RetryExecutionID)
 	assert.Equal(t, 2, exec.RetryAttemptN)
+	// idempotency_key scan-outcome guard (CR): the non-NULL path must scan
+	// the stored key into IdempotencyKey verbatim.
+	assert.Equal(t, "idem-key-exec-2", exec.IdempotencyKey)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+// TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns guards against the
+// SELECT projection in GetPlannedExecutions drifting out of sync with the
+// scanExecutionRows Scan target. When migration 000066 added idempotency_key
+// (and earlier migrations added executed_by_user_id, executed_at,
+// pre_approval_skip_reason) every execution-reading SELECT must project them,
+// or scanExecutionRows fails at runtime with "failed to scan execution" on the
+// planned-purchase list path (handler_purchases.go GetPlannedExecutions).
+//
+// The mock uses regexp query matching, so ExpectQuery requires the issued SQL
+// to contain idempotency_key; with the column missing from the projection the
+// query does not match and GetPlannedExecutions returns an error, which is what
+// this test asserts against. It fails on the pre-fix projection and passes once
+// the four trailing columns are added.
+func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) {
+	mock := newMock(t)
+	store := storeWith(mock)
+	ctx := context.Background()
+
+	recsJSON, _ := json.Marshal([]RecommendationRecord{})
+	now := time.Now().Truncate(time.Second)
+	cols := []string{
+		"plan_id", "execution_id", "status", "step_number", "scheduled_date",
+		"notification_sent", "approval_token", "recommendations",
+		"total_upfront_cost", "estimated_savings", "completed_at", "error", "expires_at",
+		"cloud_account_id", "source", "approved_by", "cancelled_by", "capacity_percent",
+		"created_by_user_id", "retry_execution_id", "retry_attempt_n",
+		"approval_token_expires_at",
+		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
+		"idempotency_key",
+	}
+	rows := pgxmock.NewRows(cols).AddRow(
+		"plan-1", "exec-1", "pending", 1, now,
+		sql.NullTime{}, "tok-123", recsJSON,
+		100.0, 200.0, sql.NullTime{}, "", sql.NullTime{},
+		nil, "", nil, nil, 100,
+		nil, nil, 0,
+		sql.NullTime{},
+		nil, sql.NullTime{}, nil,
+		"idem-key-planned",
+	)
+	// Regexp matcher: only matches if the issued SELECT projects idempotency_key.
+	mock.ExpectQuery("idempotency_key").
+		WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()).
+		WillReturnRows(rows)
+
+	execs, err := store.GetPlannedExecutions(ctx, []string{"pending"}, 10)
+	require.NoError(t, err)
+	require.Len(t, execs, 1)
+	assert.Equal(t, "idem-key-planned", execs[0].IdempotencyKey)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 

From 81ff5e31e42d8067dec38b4086650c5a1dbb57fd Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 04:25:47 -0700
Subject: [PATCH 5/7] fix(execution): guard semaphore acquire with ctx.Done()
 select (05-H3)

sem <- struct{}{} in FanOutWithConcurrency was unconditional, so a
large fan-out on a cancelled/expired context would block indefinitely
at the semaphore boundary rather than draining quickly.

Replace with a select that immediately records ctx.Err() on the
unstarted result slot when the context is done, then continues the
launch loop. The wg counter is still incremented before the select so
wg.Wait() remains correct; wg.Done() is called in the ctx.Done branch
to balance it.

Regression test: TestFanOut_ContextCancelled_BlockedSemaphore -- with
maxConcurrency=1 and a cancel fired while the first goroutine holds the
slot, the second item gets context.Canceled immediately instead of
blocking until the first goroutine releases (fails on pre-fix code).

Closes #1069 (partial).
---
 internal/execution/fanout.go      | 15 ++++++-
 internal/execution/fanout_test.go | 66 +++++++++++++++++++++++++++++++
 2 files changed, 80 insertions(+), 1 deletion(-)

diff --git a/internal/execution/fanout.go b/internal/execution/fanout.go
index 22efcdd28..6251aa30f 100644
--- a/internal/execution/fanout.go
+++ b/internal/execution/fanout.go
@@ -69,8 +69,21 @@ func FanOutWithConcurrency[T any](
 	var wg sync.WaitGroup
 
 	for i, id := range accountIDs {
+		// Acquire a semaphore slot before launching the goroutine.
+		// Use select so a cancelled/expired context is not held up by a
+		// full semaphore: if ctx is done while waiting for a slot, record
+		// ctx.Err() on the result slot and skip launching the goroutine.
+		// Without this, a large fan-out on an already-cancelled context
+		// (e.g. Lambda deadline exceeded partway through) would block
+		// indefinitely here rather than draining quickly.
 		wg.Add(1)
-		sem <- struct{}{} // acquire slot
+		select {
+		case sem <- struct{}{}:
+		case <-ctx.Done():
+			results[i] = Result[T]{AccountID: id, Err: ctx.Err()}
+			wg.Done()
+			continue
+		}
 		go func(idx int, accountID string) {
 			defer wg.Done()
 			defer func() { <-sem }() // release slot
diff --git a/internal/execution/fanout_test.go b/internal/execution/fanout_test.go
index 485888797..f8b0961e2 100644
--- a/internal/execution/fanout_test.go
+++ b/internal/execution/fanout_test.go
@@ -87,6 +87,72 @@ func TestPartition(t *testing.T) {
 // item's result slot, and does NOT propagate up and crash the whole process
 // (which would strand the surrounding purchase execution at 'approved' and
 // terminate the Lambda invocation abnormally — see #669).
+// TestFanOut_ContextCancelled_BlockedSemaphore asserts that when a context is
+// cancelled while goroutines are already occupying all semaphore slots, the
+// remaining queued items record ctx.Err() immediately rather than blocking
+// indefinitely on the semaphore (05-H3).
+//
+// Pre-fix behaviour: sem <- struct{}{} was unconditional, so a cancelled
+// context with maxConcurrency=1 and N>1 ids would block the launch loop on
+// the second item until the first goroutine released its slot -- a
+// context-deadline timeout would therefore not be respected at the semaphore
+// boundary.
+func TestFanOut_ContextCancelled_BlockedSemaphore(t *testing.T) {
+	// started (buffered=1) signals when the first goroutine holds the slot.
+	// release (buffered=1) controls when the first goroutine finishes.
+	// Both are buffered so the goroutines never block if the test is already
+	// past the receive.
+	started := make(chan struct{}, 1)
+	release := make(chan struct{}, 1)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Collect results in a separate goroutine so this test goroutine can
+	// drive the cancel/release sequencing without deadlocking on wg.Wait().
+	type outcome struct{ results []Result[string] }
+	done := make(chan outcome, 1)
+
+	go func() {
+		// maxConcurrency=1 so the second item must wait for the semaphore.
+		// The first goroutine signals started, then waits on release; during
+		// that window we cancel the context. The second item should receive
+		// ctx.Err() without waiting for the first goroutine to finish.
+		res := FanOutWithConcurrency(ctx, []string{"first", "second"},
+			func(ctx context.Context, id string) (string, error) {
+				if id == "first" {
+					started <- struct{}{} // tell the test we have the slot
+					<-release            // wait until the test says go
+					return "ok", nil
+				}
+				return "should-not-run", nil
+			}, 1)
+		done <- outcome{res}
+	}()
+
+	// Wait until the first goroutine holds the semaphore, then cancel.
+	<-started
+	cancel()
+	release <- struct{}{} // let the first goroutine finish so FanOut can return
+
+	o := <-done
+	results := o.results
+
+	require.Len(t, results, 2)
+	byID := make(map[string]Result[string], len(results))
+	for _, r := range results {
+		byID[r.AccountID] = r
+	}
+
+	// The first item completes normally (it already had the slot before cancel).
+	assert.NoError(t, byID["first"].Err)
+	assert.Equal(t, "ok", byID["first"].Value)
+
+	// The second item must carry context.Canceled, not block or succeed.
+	require.Error(t, byID["second"].Err)
+	assert.ErrorIs(t, byID["second"].Err, context.Canceled)
+}
+
 func TestFanOut_PanicInFn(t *testing.T) {
 	ids := []string{"a", "panic-me", "c"}
 	results := FanOut(context.Background(), ids, func(ctx context.Context, id string) (string, error) {

From c6cb85b76a3bc84248d6795579b72c8aac4fd270 Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 04:27:31 -0700
Subject: [PATCH 6/7] fix(scheduler): allocate fresh output slice in
 filter/suppression funcs (05-M1)

applySuppressionIndex and filterRecsByResolvedConfigs both wrote
  out := recs[:0]
which aliases the caller's backing array. Any caller that holds a
reference to the original slice would see elements silently overwritten
on the first appends into out.

Replace with make([]config.RecommendationRecord, 0, len(recs)) so the
output is always an independent allocation.

Regression tests: TestApplySuppressionIndex_DoesNotMutateCallerSlice
and TestFilterRecsByResolvedConfigs_DoesNotMutateCallerSlice snapshot
the original IDs, call the functions, and assert the original slice is
unchanged (fail on pre-fix code).

Closes #1069 (partial).
---
 internal/scheduler/scheduler.go               |  4 +-
 internal/scheduler/scheduler_overrides.go     |  4 +-
 .../scheduler/scheduler_overrides_test.go     | 38 ++++++++++++++++++
 .../scheduler/scheduler_suppressions_test.go  | 39 +++++++++++++++++++
 4 files changed, 83 insertions(+), 2 deletions(-)

diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go
index 4bba308d9..1db9960da 100644
--- a/internal/scheduler/scheduler.go
+++ b/internal/scheduler/scheduler.go
@@ -1141,7 +1141,9 @@ func aggregateSuppressions(sups []config.PurchaseSuppression) map[suppressionKey
 // count from each rec, drops recs that go to 0 or below, and
 // annotates the survivors with the badge fields.
 func applySuppressionIndex(recs []config.RecommendationRecord, index map[suppressionKey]*suppressionAgg) []config.RecommendationRecord {
-	out := recs[:0]
+	// Allocate a fresh backing array so callers that hold a reference to
+	// the original recs slice do not see mutations (05-M1).
+	out := make([]config.RecommendationRecord, 0, len(recs))
 	for _, rec := range recs {
 		accountID := ""
 		if rec.CloudAccountID != nil {
diff --git a/internal/scheduler/scheduler_overrides.go b/internal/scheduler/scheduler_overrides.go
index 468be826b..fb61c8b25 100644
--- a/internal/scheduler/scheduler_overrides.go
+++ b/internal/scheduler/scheduler_overrides.go
@@ -52,7 +52,9 @@ func filterRecsByResolvedConfigs(
 	recs []config.RecommendationRecord,
 	resolved map[string]*config.ServiceConfig,
 ) []config.RecommendationRecord {
-	out := recs[:0]
+	// Allocate a fresh backing array so callers that hold a reference to
+	// the original recs slice do not see mutations (05-M1).
+	out := make([]config.RecommendationRecord, 0, len(recs))
 	for i := range recs {
 		rec := recs[i]
 		if rec.CloudAccountID == nil {
diff --git a/internal/scheduler/scheduler_overrides_test.go b/internal/scheduler/scheduler_overrides_test.go
index 3c0d5564c..9180e7cf0 100644
--- a/internal/scheduler/scheduler_overrides_test.go
+++ b/internal/scheduler/scheduler_overrides_test.go
@@ -286,6 +286,44 @@ func TestApplyAccountOverrides_LookupError_PassesThrough(t *testing.T) {
 	assert.Len(t, recs, 1, "un-filtered list returned on lookup failure")
 }
 
+// TestFilterRecsByResolvedConfigs_DoesNotMutateCallerSlice asserts that
+// filterRecsByResolvedConfigs allocates a fresh output slice instead of
+// aliasing the caller's backing array via recs[:0] (05-M1).
+//
+// Pre-fix: out := recs[:0] shared the backing array, so any append during
+// filtering overwrote elements the caller still expected at their original
+// positions. The fix allocates make([]T, 0, len(recs)).
+func TestFilterRecsByResolvedConfigs_DoesNotMutateCallerSlice(t *testing.T) {
+	// Resolve a ServiceConfig that disables the second rec.
+	enabled := &config.ServiceConfig{Provider: "aws", Service: "rds", Enabled: true}
+	disabled := &config.ServiceConfig{Provider: "aws", Service: "rds", Enabled: false}
+	accA := "acct-A"
+	accB := "acct-B"
+	recs := []config.RecommendationRecord{
+		{ID: "keep", Provider: "aws", Service: "rds", Region: "us-east-1",
+			ResourceType: "db.t3.medium", Count: 1, CloudAccountID: &accA},
+		{ID: "drop", Provider: "aws", Service: "rds", Region: "us-east-1",
+			ResourceType: "db.t3.medium", Count: 1, CloudAccountID: &accB},
+	}
+	// Keep a snapshot of the original IDs.
+	originalIDs := []string{recs[0].ID, recs[1].ID}
+
+	resolved := map[string]*config.ServiceConfig{
+		config.AccountConfigKey(accA, "aws", "rds"): enabled,
+		config.AccountConfigKey(accB, "aws", "rds"): disabled,
+	}
+	out := filterRecsByResolvedConfigs(recs, resolved)
+
+	// Only the enabled rec survives.
+	require.Len(t, out, 1)
+	assert.Equal(t, "keep", out[0].ID)
+
+	// The original slice must not be mutated.
+	require.Len(t, recs, 2, "original slice length unchanged")
+	assert.Equal(t, originalIDs[0], recs[0].ID, "recs[0] must not be overwritten")
+	assert.Equal(t, originalIDs[1], recs[1].ID, "recs[1] must not be overwritten")
+}
+
 func TestApplyAccountOverrides_OverrideLookupError_PassesThrough(t *testing.T) {
 	// Mirrors the global-config lookup failure contract: if the per-account
 	// override lookup fails, the page should over-show rather than blank.
diff --git a/internal/scheduler/scheduler_suppressions_test.go b/internal/scheduler/scheduler_suppressions_test.go
index 694de959d..9d64411bc 100644
--- a/internal/scheduler/scheduler_suppressions_test.go
+++ b/internal/scheduler/scheduler_suppressions_test.go
@@ -197,5 +197,44 @@ func TestApplySuppressions_NilAccountIDNormalised(t *testing.T) {
 	assert.Equal(t, 3, recs[0].Count, "5 - 2 = 3 (nil account matched to empty-string suppression)")
 }
 
+// TestApplySuppressionIndex_DoesNotMutateCallerSlice asserts that
+// applySuppressionIndex allocates a fresh output slice and never writes back
+// into the caller's backing array (05-M1).
+//
+// Pre-fix: out := recs[:0] shared the caller's backing array, so any append
+// inside the function would overwrite elements that the caller still expected
+// at their original positions. The fix allocates make([]T, 0, len(recs)).
+func TestApplySuppressionIndex_DoesNotMutateCallerSlice(t *testing.T) {
+	future := time.Now().Add(24 * time.Hour)
+	// Two recs; only the second is suppressed and therefore dropped.
+	recs := []config.RecommendationRecord{
+		{ID: "keep", Provider: "aws", Service: "ec2", Region: "us-east-1",
+			ResourceType: "t4g.nano", Count: 5, CloudAccountID: strPtr("acct-1")},
+		{ID: "drop", Provider: "aws", Service: "ec2", Region: "us-east-2",
+			ResourceType: "t4g.nano", Count: 5, CloudAccountID: strPtr("acct-1")},
+	}
+	// Keep a copy of the original IDs so we can detect mutations.
+	originalIDs := []string{recs[0].ID, recs[1].ID}
+
+	sups := []config.PurchaseSuppression{
+		{ExecutionID: "e1", AccountID: "acct-1", Provider: "aws",
+			Service: "ec2", Region: "us-east-2", ResourceType: "t4g.nano",
+			SuppressedCount: 5, ExpiresAt: future},
+	}
+	index := aggregateSuppressions(sups)
+	out := applySuppressionIndex(recs, index)
+
+	// The survivor set must contain only the non-suppressed rec.
+	require.Len(t, out, 1)
+	assert.Equal(t, "keep", out[0].ID)
+
+	// The original slice must be unchanged: both elements at their
+	// original positions with the original IDs (pre-fix: recs[1].ID
+	// would be overwritten by "keep" since out alias the same array).
+	require.Len(t, recs, 2, "original slice length must not change")
+	assert.Equal(t, originalIDs[0], recs[0].ID, "recs[0] must not be mutated")
+	assert.Equal(t, originalIDs[1], recs[1].ID, "recs[1] must not be mutated")
+}
+
 // Suppress unused-import warning in case this is the only file that imports pgx.
 var _ pgx.Tx = (pgx.Tx)(nil)

From d74e045364a9e3deba1eeb0f72ac1a0ca13fe114 Mon Sep 17 00:00:00 2001
From: Cristian Magherusan-Stanciu <cristi.magherusan@gmail.com>
Date: Sun, 7 Jun 2026 04:31:19 -0700
Subject: [PATCH 7/7] fix(purchases): robustness cleanup from FOLD-1037 review
 (05-L1/L2/L3/M4/N1/N2/N3)

05-M4 (commitmentopts): add Warn log when provider probe data exists but
the requested service is absent, so operators see misconfigured service
names or probe gaps without blocking the plan save.

05-L1 (audit gap marker): prefix recordHistoryAuditGap's exec.Error note
with historyAuditGapPrefix ("history_write_failed") + commitment ID so
log queries can filter on the prefix programmatically instead of grepping
free-form prose.

05-L2 (getAWSAccountID): change signature from (string) to (string, error).
Returns an error on all failure paths instead of the "unknown" sentinel;
the sole caller logs a Warn and proceeds with an empty account ID, so
the caller decides the fallback rather than the STS helper.
Existing tests updated to assert on (id, err) pairs.

05-L3 (normalizePurchaseSource): add explanatory comment documenting the
"proceed untagged" decision and why failing the rec over a tag-only field
would be a worse outcome than an absent tag.

05-N1 (reaper _ = now): wire `now` into the sweep log line (RFC3339
timestamp) so the injected clock is actually used and the `_ = now`
suppressors can be removed.

05-N2 (aggregator contract): add a comment before the
aggregatePurchaseOutcomes call documenting the single-threaded-aggregator
invariant so it is visible at the call site, not just on the function.

05-N3 (sweep-threshold doc): expand staleApprovedThreshold's godoc to
cross-reference DefaultReapAfter in reaper.go and explain why the two
thresholds differ.

Closes #1069 (partial).
---
 internal/commitmentopts/service.go  | 12 +++++--
 internal/purchase/execution.go      | 55 ++++++++++++++++++++++-------
 internal/purchase/execution_test.go | 15 ++++----
 internal/purchase/manager.go        | 20 ++++++++---
 internal/purchase/reaper.go         | 10 +++---
 5 files changed, 81 insertions(+), 31 deletions(-)

diff --git a/internal/commitmentopts/service.go b/internal/commitmentopts/service.go
index d85ce9666..a290f7cf6 100644
--- a/internal/commitmentopts/service.go
+++ b/internal/commitmentopts/service.go
@@ -162,9 +162,15 @@ func (s *Service) Validate(ctx context.Context, provider, service string, term i
 	}
 	combos, ok := byService[service]
 	if !ok {
-		// We have data for this provider but not this service. The
-		// service is not commitment-capable in our probe set (e.g.
-		// Savings Plans) — permissive fallback.
+		// We have probe data for this provider but not for this service.
+		// This typically means the service is not commitment-capable per
+		// the probe (e.g. Savings Plans has no per-service offering list)
+		// or it is a new service not yet covered by the probe set.
+		// Log at Warn so operators can detect misconfigured service names
+		// or probe gaps without blocking the plan save (05-M4). The
+		// frontend's hardcoded rules are the primary user-facing gate.
+		logging.Warnf("commitmentopts: provider %q known in probe data but service %q absent; treating as valid (term=%d payment=%q)",
+			provider, service, term, payment)
 		return true, nil
 	}
 	for _, c := range combos {
diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go
index 0cdce9c93..8ab258a6d 100644
--- a/internal/purchase/execution.go
+++ b/internal/purchase/execution.go
@@ -86,7 +86,16 @@ func (m *Manager) executeSingleAccount(ctx context.Context, exec *config.Purchas
 	// execution where credentials are inherited from the host).
 	accountID := targetAccountID
 	if accountID == "" {
-		accountID = m.getAWSAccountID(ctx)
+		// Fall back to the ambient AWS STS identity for logging/tagging.
+		// getAWSAccountID returns ("", error) on failure so the caller
+		// can substitute a sentinel rather than baking "unknown" into the
+		// purchase history record (05-L2).
+		if id, err := m.getAWSAccountID(ctx); err != nil {
+			logging.Warnf("purchase[%s]: could not resolve ambient AWS account ID: %v; proceeding with empty account ID",
+				exec.ExecutionID, err)
+		} else {
+			accountID = id
+		}
 	}
 
 	totalSavings, totalUpfront, purchaseErrors := m.processPurchaseRecommendations(ctx, exec, plan, accountID, provCfg)
@@ -533,6 +542,11 @@ func (m *Manager) processPurchaseRecommendations(ctx context.Context, exec *conf
 			return recPurchaseOutcome{index: i, purchase: purchaseResult, err: err}, nil
 		}, getMaxAccountParallelism())
 
+	// aggregatePurchaseOutcomes is intentionally single-threaded: the fan-out
+	// results are collected above, and the aggregator walks them serially so
+	// there are no concurrent writes to totals, exec.Recommendations, or
+	// purchaseErrors (05-N2). Do NOT move the aggregation inside the FanOut
+	// closure or run it concurrently with the fan-out.
 	return m.aggregatePurchaseOutcomes(ctx, exec, plan, accountID, results)
 }
 
@@ -597,12 +611,23 @@ func (m *Manager) aggregatePurchaseOutcomes(ctx context.Context, exec *config.Pu
 // History view (which synthesises completed executions that carry an Error).
 // Appends rather than overwrites so multiple failed history writes within one
 // execution are all recorded.
+// historyAuditGapPrefix is the structured prefix stamped on exec.Error by
+// recordHistoryAuditGap. Using a fixed prefix makes these entries queryable
+// in CloudWatch Insights / log aggregators without grepping free-form text
+// (05-L1). The prefix is intentionally stable; changing it invalidates any
+// existing dashboards or alerts that pattern-match on it.
+const historyAuditGapPrefix = "history_write_failed"
+
 func recordHistoryAuditGap(exec *config.PurchaseExecution, commitmentID string, histErr error) {
 	// histErr is a raw persistence error: log it server-side for diagnosis but
 	// keep it out of exec.Error, which is surfaced to clients via the history
 	// status_description. Only a generic, user-safe note reaches the UI.
 	logging.Errorf("history audit gap for commitment %s: %v", commitmentID, histErr)
-	note := fmt.Sprintf("commitment %s purchased but its history record failed to save", commitmentID)
+	// Structured marker: historyAuditGapPrefix + ":" + commitmentID so log
+	// queries can filter on "history_write_failed" and operators can correlate
+	// the commitment ID to the cloud provider's record without parsing prose.
+	note := fmt.Sprintf("%s: commitment %s purchased but its history record failed to save",
+		historyAuditGapPrefix, commitmentID)
 	exec.Error = appendErrNote(exec.Error, note)
 }
 
@@ -644,6 +669,12 @@ func appendErrNote(existing, note string) string {
 // the allowed whitelist; an unexpected value (DB tampering, future
 // code path) is dropped to "" rather than fed onto a cloud commitment
 // where it would be expensive to retract.
+//
+// On invalid source the purchase PROCEEDS UNTAGGED (05-L3 decision):
+// failing the rec over a tag-only field would abort a successful cloud
+// purchase, which is a worse outcome than a missing tag. Input
+// validation at the API write boundary (exec.Source on save) is the
+// correct gate; this fallback is last-resort defence-in-depth.
 func (m *Manager) normalizePurchaseSource(exec *config.PurchaseExecution) string {
 	source := exec.Source
 	if source == "" {
@@ -1031,24 +1062,24 @@ func (m *Manager) updatePlanProgress(ctx context.Context, planID string) error {
 	return m.config.UpdatePurchasePlan(ctx, plan)
 }
 
-// getAWSAccountID retrieves the current AWS account ID using STS
-func (m *Manager) getAWSAccountID(ctx context.Context) string {
+// getAWSAccountID retrieves the current AWS account ID using STS.
+// It returns ("", error) on all failure paths so the caller decides how to
+// handle the absent ID (log a sentinel, skip tagging, etc.) rather than
+// silently recording "unknown" in the purchase history row (05-L2).
+func (m *Manager) getAWSAccountID(ctx context.Context) (string, error) {
 	if m.stsClient == nil {
-		logging.Debug("STS client not configured, using 'unknown' as account ID")
-		return "unknown"
+		return "", fmt.Errorf("STS client not configured")
 	}
 
 	result, err := m.stsClient.GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{})
 	if err != nil {
-		logging.Warnf("Failed to get AWS account ID: %v", err)
-		return "unknown"
+		return "", fmt.Errorf("STS GetCallerIdentity: %w", err)
 	}
 
-	if result.Account != nil {
-		return *result.Account
+	if result.Account == nil {
+		return "", fmt.Errorf("STS returned nil Account")
 	}
-
-	return "unknown"
+	return *result.Account, nil
 }
 
 // applyEngineFallback backfills the Engine field on DB / Cache Details
diff --git a/internal/purchase/execution_test.go b/internal/purchase/execution_test.go
index 620876567..310daf894 100644
--- a/internal/purchase/execution_test.go
+++ b/internal/purchase/execution_test.go
@@ -433,7 +433,8 @@ func TestManager_GetAWSAccountID_Success(t *testing.T) {
 		stsClient: mockSTS,
 	}
 
-	accountID := manager.getAWSAccountID(ctx)
+	accountID, err := manager.getAWSAccountID(ctx)
+	require.NoError(t, err)
 	assert.Equal(t, "987654321098", accountID)
 
 	mockSTS.AssertExpectations(t)
@@ -446,8 +447,8 @@ func TestManager_GetAWSAccountID_NoClient(t *testing.T) {
 		stsClient: nil, // No STS client configured
 	}
 
-	accountID := manager.getAWSAccountID(ctx)
-	assert.Equal(t, "unknown", accountID)
+	_, err := manager.getAWSAccountID(ctx)
+	require.Error(t, err, "nil STS client must return an error (not the 'unknown' sentinel)")
 }
 
 func TestManager_GetAWSAccountID_Error(t *testing.T) {
@@ -460,8 +461,8 @@ func TestManager_GetAWSAccountID_Error(t *testing.T) {
 		stsClient: mockSTS,
 	}
 
-	accountID := manager.getAWSAccountID(ctx)
-	assert.Equal(t, "unknown", accountID)
+	_, err := manager.getAWSAccountID(ctx)
+	require.Error(t, err, "STS failure must return an error")
 
 	mockSTS.AssertExpectations(t)
 }
@@ -478,8 +479,8 @@ func TestManager_GetAWSAccountID_NilAccount(t *testing.T) {
 		stsClient: mockSTS,
 	}
 
-	accountID := manager.getAWSAccountID(ctx)
-	assert.Equal(t, "unknown", accountID)
+	_, err := manager.getAWSAccountID(ctx)
+	require.Error(t, err, "nil Account in STS response must return an error")
 
 	mockSTS.AssertExpectations(t)
 }
diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go
index d783d299a..d52d37ea7 100644
--- a/internal/purchase/manager.go
+++ b/internal/purchase/manager.go
@@ -85,11 +85,21 @@ type ProcessResult struct {
 }
 
 // staleApprovedThreshold is how long an execution may sit in the "approved"
-// status before the recovery sweep treats it as stranded (issue #632). It must
-// be comfortably larger than the longest possible synchronous purchase run so a
-// legitimately in-flight execution is never failed out from under itself. The
-// purchase Lambda timeout is 60s; 15min (matching the RI-exchange stale-sweep
-// threshold in pkg/exchange) leaves a wide safety margin.
+// status before the recovery sweep in ProcessScheduledPurchases treats it as
+// stranded (issue #632). It must be comfortably larger than the longest
+// possible synchronous purchase run so a legitimately in-flight execution is
+// never failed out from under itself. The purchase Lambda timeout is 60s;
+// 15min (matching the RI-exchange stale-sweep threshold in pkg/exchange)
+// leaves a wide safety margin.
+//
+// Note: the purchase.Reaper (reaper.go) uses DefaultReapAfter (10m) to cover
+// both "approved" and "running" rows via an atomic CAS. These two thresholds
+// serve different sweep paths and the 15m vs 10m difference is intentional:
+// the reaper is CAS-protected and can safely reap "running" rows (the real
+// executor wins the CAS race if it is still alive), while this legacy sweep
+// only targets "approved" rows from the cron path and is deliberately
+// conservative (05-N3). A future consolidation should align both under a
+// single env-configurable threshold.
 const staleApprovedThreshold = 15 * time.Minute
 
 // NotificationResult holds the result of sending notifications
diff --git a/internal/purchase/reaper.go b/internal/purchase/reaper.go
index 64c589752..9312a34de 100644
--- a/internal/purchase/reaper.go
+++ b/internal/purchase/reaper.go
@@ -155,16 +155,20 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r
 	// Best-effort age estimate. The store does not currently return
 	// updated_at on the PurchaseExecution struct (issue #678 may add it
 	// later), so we lower-bound the age at reapAfter — the SELECT
-	// guarantees updated_at < NOW() - reapAfter, so the real age is at
+	// guarantees updated_at < now - reapAfter, so the real age is at
 	// least reapAfter. Rounded to whole minutes for the canonical
 	// message; the WARN log carries the same lower bound.
+	// `now` is injected (rather than time.Now()) so tests can use a
+	// fixed clock and all rows in one sweep share the same reference
+	// instant (05-N1).
 	age := reapAfter
 	ageMinutes := int(age.Round(time.Minute) / time.Minute)
 	if ageMinutes < 1 {
 		ageMinutes = 1
 	}
 
-	logging.Warnf("purchase reaper: reaping execution %s (status=%s, age>=%dm)", exec.ExecutionID, prevStatus, ageMinutes)
+	logging.Warnf("purchase reaper: reaping execution %s (status=%s, age>=%dm sweep_at=%s)",
+		exec.ExecutionID, prevStatus, ageMinutes, now.UTC().Format(time.RFC3339))
 
 	transitioned, err := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, stuckStatuses, failedStatus)
 	if err != nil {
@@ -219,10 +223,8 @@ func (m *Manager) reapOne(ctx context.Context, exec *config.PurchaseExecution, r
 		// error-message annotation failed. Otherwise Reaped would
 		// undercount real recoveries.
 		result.Reaped++
-		_ = now // kept in signature for future deterministic-clock injection
 		return
 	}
 
 	result.Reaped++
-	_ = now
 }