feat(ladder): preserve tranche amount provenance

internal/config/ladder_provenance.go

@@ -0,0 +1,41 @@
+package config
+
+import (
+	"fmt"
+	"math/big"
+)
+
+type LadderAmountBasis string
+
+const (
+	LadderAmountOnDemand   LadderAmountBasis = "on_demand_equivalent"
+	LadderAmountCommitment LadderAmountBasis = "commitment"
+)
+
+func trancheCommitmentRate(tr *LadderTrancheDB) (*string, error) {
+	if tr.AmountBasis == nil {
+		if tr.CommitmentRate != nil {
+			return nil, fmt.Errorf("commitment rate requires a known amount basis")
+		}
+		return nil, nil
+	}
+	switch *tr.AmountBasis {
+	case LadderAmountOnDemand:
+		if tr.CommitmentRate != nil {
+			return nil, fmt.Errorf("on-demand-equivalent amount must not have a commitment rate")
+		}
+		return nil, nil
+	case LadderAmountCommitment:
+		if tr.CommitmentRate == nil {
+			return nil, fmt.Errorf("commitment amount requires its stored rate")
+		}
+		rate, ok := new(big.Rat).SetString(*tr.CommitmentRate)
+		if !ok || rate.Sign() <= 0 || rate.Cmp(big.NewRat(1, 1)) > 0 {
+			return nil, fmt.Errorf("stored commitment rate must be a rational in (0, 1]")
+		}
+		canonical := rate.RatString()
+		return &canonical, nil
+	default:
+		return nil, fmt.Errorf("unknown ladder amount basis %q", *tr.AmountBasis)
+	}
+}

internal/config/store_postgres_ladder.go

@@ -293,12 +293,16 @@ func insertLadderTranchesTx(ctx context.Context, tx pgx.Tx, tranches []LadderTra
 		if tr.ID == "" {
 			return fmt.Errorf("ladder_tranche at index %d has an empty ID; every tranche must carry a unique non-empty ID", i)
 		}
-		_, err := tx.Exec(ctx, `
+		rate, err := trancheCommitmentRate(tr)
+		if err != nil {
+			return fmt.Errorf("ladder_tranche %s: %w", tr.ID, err)
+		}
+		_, err = tx.Exec(ctx, `
 			INSERT INTO ladder_tranches (
 				id, config_id, run_id, layer_type, amount_usd_hr,
 				term, payment_option, scheduled_date, status, execution_id,
-				settled_at, created_at
-			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW())
+				settled_at, amount_basis, commitment_rate, created_at
+			) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, NOW())
 		`,
 			tr.ID,
 			tr.ConfigID,
@@ -311,6 +315,8 @@ func insertLadderTranchesTx(ctx context.Context, tx pgx.Tx, tranches []LadderTra
 			string(tr.Status),
 			tr.ExecutionID,
 			tr.SettledAt,
+			tr.AmountBasis,
+			rate,
 		)
 		if err != nil {
 			return fmt.Errorf("failed to insert ladder_tranche id=%s run_id=%v: %w", tr.ID, tr.RunID, err)

internal/config/store_postgres_ladder_lifecycle.go

@@ -145,7 +145,7 @@ func (s *PostgresStore) ClaimDueLadderTranches(
 func selectDueLadderTranchesTx(ctx context.Context, tx pgx.Tx, configID string, asOf time.Time, limit int) ([]LadderTrancheDB, error) {
 	rows, err := tx.Query(ctx, `
 		SELECT id, config_id, run_id, layer_type, amount_usd_hr, term,
-		       payment_option, scheduled_date, status, settled_at, created_at
+		       payment_option, scheduled_date, status, settled_at, created_at, amount_basis, commitment_rate
 		FROM ladder_tranches
 		WHERE config_id = $1
 		  AND status = 'scheduled'
@@ -164,9 +164,12 @@ func selectDueLadderTranchesTx(ctx context.Context, tx pgx.Tx, configID string,
 	for rows.Next() {
 		var tr LadderTrancheDB
 		if err := rows.Scan(&tr.ID, &tr.ConfigID, &tr.RunID, &tr.LayerType, &tr.AmountUSDHr, &tr.Term,
-			&tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.SettledAt, &tr.CreatedAt); err != nil {
+			&tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.SettledAt, &tr.CreatedAt, &tr.AmountBasis, &tr.CommitmentRate); err != nil {
 			return nil, fmt.Errorf("failed to scan due ladder_tranche: %w", err)
 		}
+		if _, err := trancheCommitmentRate(&tr); err != nil {
+			return nil, fmt.Errorf("invalid due ladder_tranche %s: %w", tr.ID, err)
+		}
 		out = append(out, tr)
 	}
 	if err := rows.Err(); err != nil {

internal/config/store_postgres_ladder_reconcile.go

@@ -38,7 +38,7 @@ func (s *PostgresStore) ListFiredLadderTranches(ctx context.Context, limit int)
 	}
 	rows, err := s.db.Query(ctx, `
 		SELECT id, config_id, run_id, execution_id, layer_type, amount_usd_hr, term,
-		       payment_option, scheduled_date, status, settled_at, created_at
+		       payment_option, scheduled_date, status, settled_at, created_at, amount_basis, commitment_rate
 		FROM ladder_tranches
 		WHERE status = 'fired' AND execution_id IS NOT NULL
 		ORDER BY scheduled_date, id
@@ -52,9 +52,12 @@ func (s *PostgresStore) ListFiredLadderTranches(ctx context.Context, limit int)
 	for rows.Next() {
 		var tr LadderTrancheDB
 		if err := rows.Scan(&tr.ID, &tr.ConfigID, &tr.RunID, &tr.ExecutionID, &tr.LayerType, &tr.AmountUSDHr, &tr.Term,
-			&tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.SettledAt, &tr.CreatedAt); err != nil {
+			&tr.PaymentOption, &tr.ScheduledDate, &tr.Status, &tr.SettledAt, &tr.CreatedAt, &tr.AmountBasis, &tr.CommitmentRate); err != nil {
 			return nil, fmt.Errorf("failed to scan fired ladder_tranche: %w", err)
 		}
+		if _, err := trancheCommitmentRate(&tr); err != nil {
+			return nil, fmt.Errorf("invalid fired ladder_tranche %s: %w", tr.ID, err)
+		}
 		out = append(out, tr)
 	}
 	if err := rows.Err(); err != nil {

internal/config/store_postgres_ladder_test.go

@@ -22,17 +22,95 @@ package config
 
 import (
 	"context"
+	"encoding/json"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/LeanerCloud/cloud-commitments-go/pkg/ladder"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func TestLadderTrancheProvenance_InsertClaimAndReconcile(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+	store := setupLadderStore(ctx, t)
+	cfgID := seedLadderConfig(ctx, t, store)
+	asOf := time.Now().UTC().Truncate(time.Microsecond)
+	od, native, rate := LadderAmountOnDemand, LadderAmountCommitment, "2/6"
+	rows := []LadderTrancheDB{
+		lifecycleTranche(cfgID, 12.5, ladder.TrancheStatusScheduled, asOf.Add(-time.Hour)),
+		lifecycleTranche(cfgID, 8, ladder.TrancheStatusScheduled, asOf.Add(-time.Minute)),
+		lifecycleTranche(cfgID, 3, ladder.TrancheStatusScheduled, asOf),
+	}
+	rows[1].AmountBasis = &od
+	rows[2].AmountBasis, rows[2].CommitmentRate = &native, &rate
+	require.NoError(t, store.SaveLadderTranches(ctx, rows))
+	require.Equal(t, "2/6", rate, "canonical storage must not mutate caller metadata")
+	want := make(map[string]*LadderTrancheDB)
+	for i := range rows {
+		want[rows[i].ID] = &rows[i]
+	}
+	check := func(tr *LadderTrancheDB) {
+		t.Helper()
+		original := want[tr.ID]
+		require.NotNil(t, original)
+		require.Equal(t, original.AmountBasis, tr.AmountBasis)
+		require.Equal(t, original.AmountUSDHr, tr.AmountUSDHr)
+		require.Equal(t, original.ConfigID, tr.ConfigID)
+		require.Equal(t, original.RunID, tr.RunID)
+		require.Equal(t, original.LayerType, tr.LayerType)
+		require.Equal(t, original.Term, tr.Term)
+		require.Equal(t, original.PaymentOption, tr.PaymentOption)
+		require.True(t, original.ScheduledDate.Equal(tr.ScheduledDate))
+		if original.CommitmentRate == nil {
+			require.Nil(t, tr.CommitmentRate)
+		} else {
+			require.NotNil(t, tr.CommitmentRate)
+			require.Equal(t, "1/3", *tr.CommitmentRate)
+		}
+	}
+	var calls atomic.Int64
+	create := newExecCreator(ctx, t, store, &calls)
+	claimed, err := store.ClaimDueLadderTranches(ctx, cfgID, asOf, 10, func(ctx context.Context, tx pgx.Tx, tr LadderTrancheDB) (string, error) {
+		check(&tr)
+		require.Equal(t, ladder.TrancheStatusScheduled, tr.Status)
+		require.Nil(t, tr.ExecutionID)
+		return create(ctx, tx, tr)
+	})
+	require.NoError(t, err)
+	require.Len(t, claimed, len(rows))
+	require.Equal(t, int64(3), calls.Load())
+	fired, err := store.ListFiredLadderTranches(ctx, 10)
+	require.NoError(t, err)
+	require.Len(t, fired, len(rows))
+	for i := range fired {
+		check(&fired[i])
+		require.Equal(t, ladder.TrancheStatusFired, fired[i].Status)
+		require.NotNil(t, fired[i].ExecutionID)
+	}
+	for _, wire := range []string{
+		`{"commitment_rate":"1/3"}`, `{"amount_basis":"on_demand_equivalent","commitment_rate":"1/3"}`,
+		`{"amount_basis":"commitment"}`, `{"amount_basis":"unknown"}`,
+		`{"amount_basis":"commitment","commitment_rate":""}`, `{"amount_basis":"commitment","commitment_rate":"0"}`,
+		`{"amount_basis":"commitment","commitment_rate":"-1"}`, `{"amount_basis":"commitment","commitment_rate":"2"}`,
+		`{"amount_basis":"commitment","commitment_rate":"NaN"}`, `{"amount_basis":"commitment","commitment_rate":"Inf"}`,
+	} {
+		bad := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf)
+		require.NoError(t, json.Unmarshal([]byte(wire), &bad))
+		first := lifecycleTranche(cfgID, 1, ladder.TrancheStatusScheduled, asOf)
+		require.Error(t, store.SaveLadderTranches(ctx, []LadderTrancheDB{first, bad}), wire)
+		var count int
+		require.NoError(t, store.db.QueryRow(ctx, `SELECT count(*) FROM ladder_tranches`).Scan(&count))
+		require.Equal(t, len(rows), count, "invalid batch must roll back its first insert")
+	}
+}
+
 // setupLadderStore starts a container, runs migrations, and returns a store.
 // It skips (not fails) only when this environment has no usable Docker daemon,
 // so the suite degrades gracefully without a container runtime. A migration that

internal/config/types.go

@@ -1328,6 +1328,9 @@ type LadderTrancheDB struct {
 	// SettledAt is stamped by TransitionLadderTrancheTx when the tranche
 	// becomes completed; a CHECK requires it on every completed row.
 	SettledAt *time.Time `json:"settled_at,omitempty"`
+	// Nil basis preserves unproven history; native rates are lossless rational text.
+	AmountBasis    *LadderAmountBasis `json:"amount_basis,omitempty"`
+	CommitmentRate *string            `json:"commitment_rate,omitempty"`
 
 	// Required fields.
 	ID            string               `json:"id"`

internal/config/types_test.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"encoding/json"
 	"math"
 	"testing"
 	"time"
@@ -9,6 +10,37 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestLadderTrancheDB_AmountProvenanceJSON(t *testing.T) {
+	t.Parallel()
+	for _, tc := range []struct {
+		name, wire  string
+		basis, rate any
+	}{
+		{"historical unknown", `{"amount_usd_hr":12.5}`, nil, nil},
+		{"explicit null unknown", `{"amount_usd_hr":12.5,"amount_basis":null,"commitment_rate":null}`, nil, nil},
+		{"verified OD", `{"amount_usd_hr":12.5,"amount_basis":"on_demand_equivalent"}`, "on_demand_equivalent", nil},
+		{"native exact rate", `{"amount_usd_hr":12.5,"amount_basis":"commitment","commitment_rate":"1/3"}`, "commitment", "1/3"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var tr LadderTrancheDB
+			require.NoError(t, json.Unmarshal([]byte(tc.wire), &tr))
+			encoded, err := json.Marshal(tr)
+			require.NoError(t, err)
+			var got map[string]any
+			require.NoError(t, json.Unmarshal(encoded, &got))
+			assert.Equal(t, 12.5, got["amount_usd_hr"])
+			assert.Equal(t, tc.basis, got["amount_basis"])
+			assert.Equal(t, tc.rate, got["commitment_rate"])
+			if tc.basis == nil {
+				assert.NotContains(t, got, "amount_basis")
+			}
+			if tc.rate == nil {
+				assert.NotContains(t, got, "commitment_rate")
+			}
+		})
+	}
+}
+
 func TestRampSchedule_GetCurrentCoverage(t *testing.T) {
 	tests := []struct {
 		schedule     RampSchedule

internal/database/postgres/migrations/000121_ladder_tranche_amount_provenance.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE ladder_tranches DROP CONSTRAINT ladder_tranche_amount_provenance;
+ALTER TABLE ladder_tranches
+    DROP COLUMN commitment_rate,
+    DROP COLUMN amount_basis;

internal/database/postgres/migrations/000121_ladder_tranche_amount_provenance.up.sql

@@ -0,0 +1,11 @@
+ALTER TABLE ladder_tranches
+    ADD COLUMN amount_basis TEXT,
+    ADD COLUMN commitment_rate TEXT;
+
+ALTER TABLE ladder_tranches ADD CONSTRAINT ladder_tranche_amount_provenance CHECK (
+    (amount_basis IS NULL AND commitment_rate IS NULL)
+    OR (amount_basis IS NOT NULL AND (
+        (amount_basis = 'on_demand_equivalent' AND commitment_rate IS NULL)
+        OR (amount_basis = 'commitment' AND commitment_rate IS NOT NULL AND commitment_rate <> '')
+    ))
+) NOT VALID;

internal/database/postgres/migrations/000121_ladder_tranche_amount_provenance_test.go

@@ -0,0 +1,65 @@
+//go:build integration
+
+package migrations_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations"
+	"github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigration_LadderTrancheAmountProvenance(t *testing.T) {
+	ctx := context.Background()
+	container := testhelpers.RequirePostgresContainer(ctx, t)
+	t.Cleanup(func() { require.NoError(t, container.Cleanup(context.Background())) })
+	pool, path := container.DB.Pool(), getMigrationsPath()
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 120))
+	var id string
+	require.NoError(t, pool.QueryRow(ctx, `INSERT INTO ladder_tranches
+ (layer_type,amount_usd_hr,term,payment_option,scheduled_date,status)
+ VALUES ('compute-sp',12.5,'1y','no_upfront',NOW(),'scheduled') RETURNING id`).Scan(&id))
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 121))
+	var validated bool
+	require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname='ladder_tranche_amount_provenance'`).Scan(&validated))
+	require.False(t, validated)
+	var amount string
+	var basis, rate *string
+	require.NoError(t, pool.QueryRow(ctx, `SELECT amount_usd_hr::text,amount_basis,commitment_rate
+ FROM ladder_tranches WHERE id=$1`, id).Scan(&amount, &basis, &rate))
+	require.Equal(t, "12.500000", amount)
+	require.Nil(t, basis)
+	require.Nil(t, rate)
+	for _, pair := range [][2]any{{"on_demand_equivalent", nil}, {"commitment", "1/3"}, {nil, nil}} {
+		_, err := pool.Exec(ctx, `UPDATE ladder_tranches SET amount_basis=$2,commitment_rate=$3 WHERE id=$1`, id, pair[0], pair[1])
+		require.NoError(t, err)
+	}
+	for _, pair := range [][2]any{{nil, "1/3"}, {"on_demand_equivalent", "1/3"}, {"commitment", nil}, {"commitment", ""}, {"unknown", nil}} {
+		_, err := pool.Exec(ctx, `UPDATE ladder_tranches SET amount_basis=$2,commitment_rate=$3 WHERE id=$1`, id, pair[0], pair[1])
+		require.Error(t, err, "invalid provenance pair %v", pair)
+	}
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 122))
+	require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname='ladder_tranche_amount_provenance'`).Scan(&validated))
+	require.True(t, validated)
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 121))
+	require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname='ladder_tranche_amount_provenance'`).Scan(&validated))
+	require.False(t, validated)
+	_, err := pool.Exec(ctx, `UPDATE ladder_tranches SET amount_basis='commitment' WHERE id=$1`, id)
+	require.Error(t, err)
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 120))
+	var columns int
+	require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM information_schema.columns
+ WHERE table_name='ladder_tranches' AND column_name IN ('amount_basis','commitment_rate')`).Scan(&columns))
+	require.Zero(t, columns)
+	require.NoError(t, pool.QueryRow(ctx, `SELECT amount_usd_hr::text FROM ladder_tranches WHERE id=$1`, id).Scan(&amount))
+	require.Equal(t, "12.500000", amount)
+	require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 122))
+	require.NoError(t, pool.QueryRow(ctx, `SELECT convalidated FROM pg_constraint WHERE conname='ladder_tranche_amount_provenance'`).Scan(&validated))
+	require.True(t, validated)
+	require.NoError(t, pool.QueryRow(ctx, `SELECT amount_usd_hr::text,amount_basis,commitment_rate FROM ladder_tranches WHERE id=$1`, id).Scan(&amount, &basis, &rate))
+	require.Equal(t, "12.500000", amount)
+	require.Nil(t, basis)
+	require.Nil(t, rate)
+}

internal/database/postgres/migrations/000122_validate_ladder_tranche_amount_provenance.down.sql

@@ -0,0 +1,8 @@
+ALTER TABLE ladder_tranches DROP CONSTRAINT ladder_tranche_amount_provenance;
+ALTER TABLE ladder_tranches ADD CONSTRAINT ladder_tranche_amount_provenance CHECK (
+    (amount_basis IS NULL AND commitment_rate IS NULL)
+    OR (amount_basis IS NOT NULL AND (
+        (amount_basis = 'on_demand_equivalent' AND commitment_rate IS NULL)
+        OR (amount_basis = 'commitment' AND commitment_rate IS NOT NULL AND commitment_rate <> '')
+    ))
+) NOT VALID;

internal/database/postgres/migrations/000122_validate_ladder_tranche_amount_provenance.up.sql

@@ -0,0 +1 @@
+ALTER TABLE ladder_tranches VALIDATE CONSTRAINT ladder_tranche_amount_provenance;

internal/server/handler_ladder.go

@@ -703,6 +703,7 @@ func buildTrancheDBRows(tranches []pkgladder.Tranche, runID string, configID *st
 		}
 		amountUSDHr, _ := r.Float64()
 		runIDCopy := runID
+		basis := config.LadderAmountOnDemand // The pinned pre-go349 SDK emits OD amounts.
 		rows = append(rows, config.LadderTrancheDB{
 			ID:            tr.ID,
 			ConfigID:      configID,
@@ -712,6 +713,7 @@ func buildTrancheDBRows(tranches []pkgladder.Tranche, runID string, configID *st
 			PaymentOption: tr.PaymentOption,
 			Status:        pkgladder.TrancheStatusScheduled,
 			AmountUSDHr:   amountUSDHr,
+			AmountBasis:   &basis,
 			ScheduledDate: tr.FireAfter,
 		})
 	}

internal/server/handler_ladder_test.go

@@ -602,6 +602,14 @@ func TestExecuteLadderRun_ImmediateStepIsTrancheAndPlanShowsOnlyIt(t *testing.T)
 	require.NoError(t, app.executeLadderRun(ctx, &dbCfg, cap, "123456789012", pkgladder.Term1Year, pkgladder.PaymentNoUpfront, now))
 
 	require.Len(t, store.savedTranches, 2)
+	for i := range store.savedTranches {
+		encoded, err := json.Marshal(&store.savedTranches[i])
+		require.NoError(t, err)
+		var persisted map[string]any
+		require.NoError(t, json.Unmarshal(encoded, &persisted))
+		assert.Equal(t, "on_demand_equivalent", persisted["amount_basis"])
+		assert.NotContains(t, persisted, "commitment_rate")
+	}
 	assert.True(t, store.savedTranches[0].ScheduledDate.Equal(now), "immediate tranche scheduled_date must be the run start")
 	assert.InDelta(t, 2.5, store.savedTranches[0].AmountUSDHr, 1e-9)
 	assert.InDelta(t, 7.5, store.savedTranches[1].AmountUSDHr, 1e-9)