feat(purchase): persist no_submission proof on failed executions (#118 PR-C1)

internal/config/store_postgres.go

@@ -1052,8 +1052,8 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 			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, scheduled_execution_at
-		) 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, $27)
+			idempotency_key, scheduled_execution_at, no_submission
+		) 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, $27, $28)
 		ON CONFLICT (execution_id) DO UPDATE SET
 			status = $3,
 			notification_sent = $6,
@@ -1074,6 +1074,9 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 			executed_at = $24,
 			pre_approval_skip_reason = $25,
 			scheduled_execution_at = $27,
+			no_submission = CASE WHEN $3 = 'failed'
+				THEN purchase_executions.no_submission OR $28
+				ELSE false END,
 			updated_at = NOW()
 	`
 
@@ -1128,6 +1131,7 @@ func (s *PostgresStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx,
 		execution.PreApprovalSkipReason,
 		execution.IdempotencyKey,
 		execution.ScheduledExecutionAt,
+		execution.NoSubmission,
 	)
 
 	if err != nil {
@@ -1229,7 +1233,7 @@ func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, execution
 		          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, scheduled_execution_at
+		          idempotency_key, scheduled_execution_at, no_submission
 	`
 
 	records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty)
@@ -1421,7 +1425,7 @@ func (s *PostgresStore) GetExecutionsByStatuses(ctx context.Context, statuses []
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		  AND NOT (status = 'completed' AND COALESCE(error, '') = '')
@@ -1533,7 +1537,7 @@ func (s *PostgresStore) GetPlannedExecutions(ctx context.Context, statuses []str
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		ORDER BY scheduled_date ASC NULLS LAST, id ASC
@@ -1558,7 +1562,7 @@ func (s *PostgresStore) GetStaleApprovedExecutions(ctx context.Context, olderTha
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status = 'approved' AND updated_at < NOW() - $1::interval
 	`
@@ -1600,7 +1604,7 @@ func (s *PostgresStore) ListStuckExecutions(ctx context.Context, statuses []stri
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status = ANY($1)
 		  AND updated_at < NOW() - $2::interval
@@ -1621,7 +1625,7 @@ func (s *PostgresStore) GetPendingExecutions(ctx context.Context) ([]PurchaseExe
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status IN ('pending', 'notified')
 		  AND (expires_at IS NULL OR expires_at > NOW())
@@ -1645,7 +1649,7 @@ func (s *PostgresStore) GetPendingExecutionsTx(ctx context.Context, tx pgx.Tx) (
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status IN ('pending', 'notified')
 		  AND (expires_at IS NULL OR expires_at > NOW())
@@ -1693,7 +1697,7 @@ func (s *PostgresStore) ListRecentSubmitsTx(ctx context.Context, tx pgx.Tx, crea
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE source = $3
 		  AND status IN (` + submitDedupeStatuses + `)
@@ -1723,7 +1727,7 @@ func (s *PostgresStore) GetExecutionByID(ctx context.Context, executionID 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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE execution_id = $1
 	`
@@ -1750,7 +1754,7 @@ func (s *PostgresStore) GetExecutionByPlanAndDate(ctx context.Context, planID st
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE plan_id = $1 AND scheduled_date = $2
 	`
@@ -1918,6 +1922,7 @@ func scanExecutionRows(rows pgx.Rows) ([]PurchaseExecution, error) {
 			&exec.PreApprovalSkipReason,
 			&idempotencyKey,
 			&scheduledExecutionAt,
+			&exec.NoSubmission,
 		)
 		if err != nil {
 			return nil, fmt.Errorf("failed to scan execution: %w", err)
@@ -1958,7 +1963,7 @@ func (s *PostgresStore) GetScheduledExecutionsDue(ctx context.Context) ([]Purcha
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE status = 'scheduled'
 		  AND scheduled_execution_at IS NOT NULL

internal/config/store_postgres_no_submission_integration_test.go

@@ -0,0 +1,132 @@
+//go:build integration
+
+package config
+
+// Real PostgreSQL tests for the persisted no_submission proof (issue #118,
+// PR-C1). No mocks: the upsert and the readers run against a migrated schema.
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newNoSubmissionExec(ctx context.Context, t *testing.T, store *PostgresStore, status string) *PurchaseExecution {
+	t.Helper()
+	var planID string
+	require.NoError(t, store.db.QueryRow(ctx, `INSERT INTO purchase_plans (name) VALUES ('c1-proof') RETURNING id`).Scan(&planID))
+	exec := &PurchaseExecution{PlanID: planID, Status: status, StepNumber: 1, ScheduledDate: time.Now()}
+	require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+	return exec
+}
+
+func storedNoSubmission(ctx context.Context, t *testing.T, store *PostgresStore, id string) bool {
+	t.Helper()
+	var v bool
+	require.NoError(t, store.db.QueryRow(ctx, `SELECT no_submission FROM purchase_executions WHERE execution_id = $1`, id).Scan(&v))
+	return v
+}
+
+// Among failed writes the proof is monotone: a stale failed copy (false)
+// upserted over a failed row that carries the proof (true) must not clear it,
+// while false -> true works. A plain overwrite would fail the last assertion.
+func TestNoSubmission_UpsertIsMonotone(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	exec := newNoSubmissionExec(ctx, t, store, "running")
+	assert.False(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "default is no proof")
+
+	stale := *exec
+	exec.Status = "failed"
+	exec.NoSubmission = true
+	require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+	assert.True(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "false -> true must persist")
+
+	stale.Status = "failed"
+	stale.NoSubmission = false
+	require.NoError(t, store.SavePurchaseExecution(ctx, &stale))
+	assert.True(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "a stale false upsert must not clear the proof")
+}
+
+// Every reader of purchase_executions returns the flag (the shared column list
+// is edited in twelve SELECT/RETURNING sites).
+func TestNoSubmission_ReadersReturnTheFlag(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	exec := newNoSubmissionExec(ctx, t, store, "running")
+	exec.Status = "failed" // the proof is only kept on a failed write
+	exec.NoSubmission = true
+	require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+
+	got, err := store.GetExecutionByID(ctx, exec.ExecutionID)
+	require.NoError(t, err)
+	assert.True(t, got.NoSubmission, "GetExecutionByID")
+
+	byPlan, err := store.GetExecutionByPlanAndDate(ctx, exec.PlanID, exec.ScheduledDate)
+	require.NoError(t, err)
+	assert.True(t, byPlan.NoSubmission, "GetExecutionByPlanAndDate")
+
+	time.Sleep(1100 * time.Millisecond) // updated_at is trigger-managed; age the row past the 1s floor
+	stuck, err := store.ListStuckExecutions(ctx, []string{"failed"}, time.Second)
+	require.NoError(t, err)
+	require.Len(t, stuck, 1)
+	assert.True(t, stuck[0].NoSubmission, "ListStuckExecutions")
+
+	failed, err := store.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"failed"}, "approved", nil)
+	require.NoError(t, err)
+	assert.True(t, failed.NoSubmission, "TransitionExecutionStatus RETURNING")
+}
+
+// Defense in depth: only a failed write may keep the proof. A completed write
+// (or any other status) over a failed+true row clears it.
+func TestNoSubmission_NonFailedWriteClearsProof(t *testing.T) {
+	ctx := context.Background()
+	for _, status := range []string{"completed", "approved"} {
+		t.Run(status, func(t *testing.T) {
+			store := setupLadderStore(ctx, t)
+			exec := newNoSubmissionExec(ctx, t, store, "running")
+			stale := *exec
+			exec.Status = "failed"
+			exec.NoSubmission = true
+			require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+			require.True(t, storedNoSubmission(ctx, t, store, exec.ExecutionID))
+
+			stale.Status = status
+			stale.NoSubmission = false
+			require.NoError(t, store.SavePurchaseExecution(ctx, &stale))
+			assert.False(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "a %s write must clear the proof", status)
+		})
+	}
+}
+
+// Replays the approve-path sequence at the store level (approvals.go:392-421):
+// the approve path keeps a stale copy of the approved row; a concurrent
+// claimAndExecute fails the row before submission (proof true); the approve
+// path's audit upsert puts the row back to approved; the approve path then
+// calls the provider and the row ends failed with outcome unknown (flag
+// false). The proof must not survive on a row that reached the provider.
+func TestNoSubmission_StaleApproveUpsertDoesNotKeepProofOnSubmittedRow(t *testing.T) {
+	ctx := context.Background()
+	store := setupLadderStore(ctx, t)
+	exec := newNoSubmissionExec(ctx, t, store, "approved")
+	staleApproved := *exec
+
+	exec.Status = "failed" // claimAndExecute: refused before submission
+	exec.NoSubmission = true
+	require.NoError(t, store.SavePurchaseExecution(ctx, exec))
+
+	require.NoError(t, store.SavePurchaseExecution(ctx, &staleApproved)) // approvals.go:412
+	var status string
+	require.NoError(t, store.db.QueryRow(ctx, `SELECT status FROM purchase_executions WHERE execution_id = $1`, exec.ExecutionID).Scan(&status))
+	require.Equal(t, "approved", status)
+	assert.False(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "proof must be cleared when the row goes back to approved")
+
+	final := staleApproved
+	final.Status = "failed" // provider called, outcome unknown: no proof
+	final.NoSubmission = false
+	require.NoError(t, store.SavePurchaseExecution(ctx, &final))
+	assert.False(t, storedNoSubmission(ctx, t, store, exec.ExecutionID), "a failed row that reached the provider must carry no proof")
+}

internal/config/store_postgres_pgxmock_test.go

@@ -658,6 +658,7 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) {
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key",
 		"scheduled_execution_at",
+		"no_submission",
 	}
 	rows := pgxmock.NewRows(cols).AddRow(
 		"plan-1", "exec-1", "pending", 1, now,
@@ -669,6 +670,7 @@ func TestPGXMock_GetExecutionByID_Success(t *testing.T) {
 		nil, sql.NullTime{}, nil,
 		nil,            // idempotency_key (NULL: legacy-row scan path, migration 000066)
 		sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path)
+		false,          // no_submission
 	)
 	mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows)
 
@@ -708,6 +710,7 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) {
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key",
 		"scheduled_execution_at",
+		"no_submission",
 	}
 	successorID := "exec-3"
 	rows := pgxmock.NewRows(cols).AddRow(
@@ -723,6 +726,7 @@ func TestPGXMock_GetExecutionByID_WithTimestamps(t *testing.T) {
 		nil, sql.NullTime{}, nil,
 		"idem-key-exec-2", // idempotency_key non-NULL: exercises the scan path
 		sql.NullTime{},    // scheduled_execution_at (NULL: not on the pre-fire delay path)
+		false,             // no_submission
 	)
 	mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows)
 
@@ -774,6 +778,7 @@ func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) {
 		"approval_token_expires_at",
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key", "scheduled_execution_at",
+		"no_submission",
 	}
 	schedAt := now.Add(2 * time.Hour)
 	rows := pgxmock.NewRows(cols).AddRow(
@@ -786,6 +791,7 @@ func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) {
 		nil, sql.NullTime{}, nil,
 		"idem-key-planned",
 		sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path)
+		false,          // no_submission
 	).AddRow(
 		"plan-1", "exec-2", "pending", 1, now,
 		sql.NullTime{}, "tok-456", recsJSON,
@@ -796,6 +802,7 @@ func TestPGXMock_GetPlannedExecutions_ProjectsAllScanColumns(t *testing.T) {
 		nil, sql.NullTime{}, nil,
 		"idem-key-delayed",
 		sql.NullTime{Time: schedAt, Valid: true}, // scheduled_execution_at populated (pre-fire delay path)
+		false,                                    // no_submission
 	)
 	// Regexp matcher: only matches if the issued SELECT projects both
 	// idempotency_key and scheduled_execution_at. The alternation forces both
@@ -854,6 +861,7 @@ func TestPGXMock_GetExecutionByID_ProjectsCoalescedCancelledBy(t *testing.T) {
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key",
 		"scheduled_execution_at",
+		"no_submission",
 	}
 	// Simulates the exact row shape the atomic-cancel paths produce: the
 	// actor lives only in canceled_by (legacy cancelled_by is NULL). What
@@ -870,6 +878,7 @@ func TestPGXMock_GetExecutionByID_ProjectsCoalescedCancelledBy(t *testing.T) {
 		nil, sql.NullTime{}, nil,
 		nil,
 		sql.NullTime{},
+		false,
 	)
 	mock.ExpectQuery(`COALESCE\(canceled_by,\s*cancelled_by\)`).
 		WithArgs(pgxmock.AnyArg()).
@@ -2562,6 +2571,7 @@ func stuckExecRow(execID, status string, scheduled time.Time) []any {
 		nil, sql.NullTime{}, nil,
 		nil,            // idempotency_key (NULL: legacy-row scan path, migration 000066)
 		sql.NullTime{}, // scheduled_execution_at (NULL: not on the pre-fire delay path)
+		false,          // no_submission
 	}
 }
 
@@ -2576,6 +2586,7 @@ func stuckExecCols() []string {
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key",
 		"scheduled_execution_at",
+		"no_submission",
 	}
 }
 
@@ -2769,6 +2780,7 @@ func TestPGXMock_TransitionExecutionStatus_ProbeHardErrorNotMappedToNotFound(t *
 		"approval_token_expires_at",
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key", "scheduled_execution_at",
+		"no_submission",
 	}
 
 	// CAS UPDATE matches zero rows (status already transitioned or row gone).
@@ -2816,6 +2828,7 @@ func TestPGXMock_GetExecutionByPlanAndDate_NotFoundWrapsErrNotFound(t *testing.T
 		"approval_token_expires_at",
 		"executed_by_user_id", "executed_at", "pre_approval_skip_reason",
 		"idempotency_key", "scheduled_execution_at",
+		"no_submission",
 	}
 	emptyRows := pgxmock.NewRows(cols)
 	mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()).WillReturnRows(emptyRows)

internal/config/store_postgres_ramp_step.go

@@ -331,7 +331,7 @@ func (s *PostgresStore) ListExecutionsForPlanStepTx(ctx context.Context, tx pgx.
 		       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, scheduled_execution_at
+		       idempotency_key, scheduled_execution_at, no_submission
 		FROM purchase_executions
 		WHERE plan_id = $1 AND step_number = $2
 		ORDER BY retry_attempt_n, execution_id`, planID, stepNumber)

internal/config/types.go

@@ -394,6 +394,14 @@ type PurchaseExecution struct {
 	// scheduler fires the actual SDK call when this timestamp is in the past.
 	// NULL on every immediate-execute row. Migration 000065.
 	ScheduledExecutionAt *time.Time `json:"scheduled_execution_at,omitempty" dynamodbav:"scheduled_execution_at,omitempty"`
+	// NoSubmission is the persisted proof that this execution failed before
+	// any cloud purchase call was made (migration 000113). Only the
+	// pre-submission failure paths set it; the reaper and every
+	// outcome-unknown failure leave it false. It is monotone: the upsert can
+	// set it but never clear it, so a stale in-memory copy cannot erase it.
+	// The ladder reconciler treats a failed execution as "nothing was bought"
+	// only when this is true.
+	NoSubmission bool `json:"no_submission,omitempty" dynamodbav:"no_submission,omitempty"`
 }
 
 // StatusCanceled is the canonical US-spelling status value new code writes.