fix(ladder): convert amend ceiling amounts by the amended tranche's basis

internal/config/store_postgres_ladder_amend.go

@@ -56,6 +56,8 @@ type lockedLadderTranche struct {
 	executionID *string
 	revision    int64
 	amount      string
+	basis       *LadderAmountBasis
+	rate        *string
 	date        time.Time
 	future      bool
 }
@@ -96,7 +98,7 @@ func amendLadderTrancheTx(ctx context.Context, tx pgx.Tx, configID, id, accountI
 	if err != nil {
 		return err
 	}
-	err = checkLadderConfigCeiling(ctx, tx, run.configID, tranche.amount, amendment.AmountUSDHr)
+	err = checkLadderConfigCeiling(ctx, tx, run.configID, tranche, amendment.AmountUSDHr)
 	if err != nil {
 		return err
 	}
@@ -140,10 +142,10 @@ func lockLadderRun(ctx context.Context, tx pgx.Tx, id, accountID, provider strin
 
 func lockLadderTranche(ctx context.Context, tx pgx.Tx, id string, run lockedLadderRun, proposed time.Time) (lockedLadderTranche, error) {
 	var t lockedLadderTranche
-	err := tx.QueryRow(ctx, `SELECT status,execution_id,revision,amount_usd_hr::text,scheduled_date,
+	err := tx.QueryRow(ctx, `SELECT status,execution_id,revision,amount_usd_hr::text,amount_basis,commitment_rate,scheduled_date,
  scheduled_date>clock_timestamp() AND $4::timestamptz>clock_timestamp()
  FROM ladder_tranches WHERE id=$1 AND config_id=$2 AND run_id=$3 FOR UPDATE`, id, run.configID, run.runID, proposed).
-		Scan(&t.status, &t.executionID, &t.revision, &t.amount, &t.date, &t.future)
+		Scan(&t.status, &t.executionID, &t.revision, &t.amount, &t.basis, &t.rate, &t.date, &t.future)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return t, ErrLadderAmendNotFound
 	}

internal/config/store_postgres_ladder_amend_ceiling.go

@@ -5,12 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"math/big"
 	"time"
 
 	"github.com/jackc/pgx/v5"
 )
 
-const ladderMicros = 1e6
+const ladderMicros = 1_000_000
 
 // ladderMaxUSDHr keeps micro-dollar sums far below int64 overflow (about 9.2e12 USD/hr).
 const ladderMaxUSDHr = 1e12
@@ -26,20 +27,22 @@ func toLadderMicros(v float64) int64 { return int64(math.Round(v * ladderMicros)
 // Committed = in-flight (fired + scheduled, all runs, with the amended row at
 // its new amount) + tranches completed after that run started, less the
 // ladderSettleSkewMargin for clock skew (their
-// commitment may not be in the run's existing_usd_hr snapshot yet).
+// commitment may not be in the run's existing_usd_hr snapshot yet). Every term
+// is netted per row in on-demand-equivalent dollars, the unit of the ceiling;
+// the amended row's old and new amounts convert by its own basis and rate.
 // Ceiling C = min(target, baseline * current coverage%) - existing, from the
 // run's snapshot (existing is gross, so C errs low) and the config's current
 // coverage. A run without a full snapshot refuses a raise. Must run inside
 // withLadderConfigLock so the sums cannot move under it.
-func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmount, newAmount string) error {
+func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID string, amended lockedLadderTranche, newAmount string) error {
 	var raise bool
-	if err := tx.QueryRow(ctx, `SELECT $1::numeric>$2::numeric`, newAmount, oldAmount).Scan(&raise); err != nil {
+	if err := tx.QueryRow(ctx, `SELECT $1::numeric>$2::numeric`, newAmount, amended.amount).Scan(&raise); err != nil {
 		return err
 	}
 	if !raise {
 		return nil
 	}
-	snap, err := loadLadderCeiling(ctx, tx, configID, oldAmount, newAmount)
+	snap, err := loadLadderCeiling(ctx, tx, configID)
 	if err != nil {
 		return err
 	}
@@ -51,57 +54,69 @@ func checkLadderConfigCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmoun
 	if err != nil {
 		return err
 	}
-	inFlight, settled, err := ladderCommittedOnDemand(ctx, tx, configID, now, ladderSettleAnchor(snap.anchor))
+	committed, err := ladderCommittedWithAmend(ctx, tx, configID, now, ladderSettleAnchor(snap.anchor), amended, newAmount)
 	if err != nil {
 		return err
 	}
-	for _, v := range []float64{inFlight, settled, snap.ceiling, snap.oldAmount, snap.newAmount} {
-		if v > ladderMaxUSDHr {
-			return fmt.Errorf("%w: amount beyond %.0f USD/hr", ErrLadderAmendInvalid, ladderMaxUSDHr)
-		}
+	if snap.ceiling > ladderMaxUSDHr {
+		return fmt.Errorf("%w: amount beyond %.0f USD/hr", ErrLadderAmendInvalid, ladderMaxUSDHr)
 	}
-	committed := toLadderMicros(inFlight) - toLadderMicros(snap.oldAmount) + toLadderMicros(snap.newAmount) + toLadderMicros(settled)
-	if committed > toLadderMicros(snap.ceiling) {
+	if committed.Cmp(big.NewRat(toLadderMicros(snap.ceiling), ladderMicros)) > 0 {
 		return ErrLadderAmendInvalid
 	}
 	return nil
 }
 
-// ladderCommittedOnDemand nets the in-flight and settled-after-anchor rows to
-// on-demand-equivalent dollars, assuming on-demand for rows with no stored basis.
-func ladderCommittedOnDemand(ctx context.Context, tx pgx.Tx, configID string, now, anchor time.Time) (inFlight, settled float64, err error) {
+// ladderCommittedWithAmend is the on-demand-equivalent committed total with the
+// amended row swapped from its old to its new amount, each converted by that
+// row's own basis and rate. The rows still hold the old amount.
+func ladderCommittedWithAmend(ctx context.Context, tx pgx.Tx, configID string, now, anchor time.Time, amended lockedLadderTranche, newAmount string) (*big.Rat, error) {
 	inFlightRows, err := inFlightLadderRows(ctx, tx, true, configID, now)
 	if err != nil {
-		return 0, 0, err
+		return nil, err
 	}
 	settledRows, err := settledLadderRows(ctx, tx, true, configID, anchor)
 	if err != nil {
-		return 0, 0, err
+		return nil, err
+	}
+	committed, err := sumLadderOnDemand(inFlightRows, settledRows)
+	if err != nil {
+		return nil, err
+	}
+	tr := assumeOnDemandIfUnknown(&LadderTrancheDB{AmountBasis: amended.basis, CommitmentRate: amended.rate})
+	oldOD, err := nettingOnDemand(amended.amount, tr)
+	if err != nil {
+		return nil, err
+	}
+	newOD, err := nettingOnDemand(newAmount, tr)
+	if err != nil {
+		return nil, err
 	}
-	if inFlight, err = LadderInFlightOnDemand(inFlightRows); err != nil {
-		return 0, 0, err
+	maxOD := new(big.Rat).SetInt64(ladderMaxUSDHr)
+	for _, v := range []*big.Rat{committed, oldOD, newOD} {
+		if v.Cmp(maxOD) > 0 {
+			return nil, fmt.Errorf("%w: amount beyond %.0f USD/hr", ErrLadderAmendInvalid, ladderMaxUSDHr)
+		}
 	}
-	settled, err = LadderInFlightOnDemand(settledRows)
-	return inFlight, settled, err
+	return committed.Sub(committed, oldOD).Add(committed, newOD), nil
 }
 
 type ladderCeilingSnapshot struct {
-	anchor                        time.Time
-	ceiling, oldAmount, newAmount float64
+	anchor  time.Time
+	ceiling float64
 }
 
 // loadLadderCeiling reads the latest non-canceled run's snapshot joined with the
 // config's current coverage. It refuses a run without a full snapshot.
-func loadLadderCeiling(ctx context.Context, tx pgx.Tx, configID, oldAmount, newAmount string) (ladderCeilingSnapshot, error) {
+func loadLadderCeiling(ctx context.Context, tx pgx.Tx, configID string) (ladderCeilingSnapshot, error) {
 	var snap ladderCeilingSnapshot
 	var hasSnapshot bool
 	err := tx.QueryRow(ctx, `SELECT r.started_at,
  r.target_usd_hr IS NOT NULL AND r.existing_usd_hr IS NOT NULL AND r.baseline_usd_hr IS NOT NULL,
- COALESCE(trunc(LEAST(r.target_usd_hr,r.baseline_usd_hr*c.target_coverage/100)-r.existing_usd_hr,6),0)::float8,
- $2::numeric::float8,$3::numeric::float8
+ COALESCE(trunc(LEAST(r.target_usd_hr,r.baseline_usd_hr*c.target_coverage/100)-r.existing_usd_hr,6),0)::float8
  FROM ladder_runs r JOIN ladder_configs c ON c.id=r.config_id
  WHERE r.config_id=$1 AND r.status<>'cancelled' ORDER BY r.started_at DESC,r.created_at DESC,r.id DESC LIMIT 1`,
-		configID, oldAmount, newAmount).Scan(&snap.anchor, &hasSnapshot, &snap.ceiling, &snap.oldAmount, &snap.newAmount)
+		configID).Scan(&snap.anchor, &hasSnapshot, &snap.ceiling)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return snap, fmt.Errorf("%w: config has no non-canceled run", ErrLadderAmendInvalid)
 	}

internal/config/store_postgres_ladder_basis_integration_test.go

@@ -29,6 +29,37 @@ func (f *ceilingFixture) setBasis(id, basis, rate string) {
 	require.NoError(f.t, err)
 }
 
+func TestLadderAmendCeiling_RaiseOnCommitmentRowNetsInOnDemand(t *testing.T) {
+	f := newCeilingFixture(t)
+	_, ids := f.plan(fptr(10), fptr(0), fptr(100), 4)
+	f.setBasis(ids[0], "commitment", "1/2")
+	require.ErrorIs(t, f.amend(ids[0], 6), ErrLadderAmendInvalid, "6 commitment at rate 1/2 is 12 on-demand against a ceiling of 10")
+	require.NoError(t, f.amend(ids[0], 5), "5 commitment at rate 1/2 is exactly the ceiling of 10")
+}
+
+func TestLadderAmendCeiling_NullBasisIsAssumedOnDemand(t *testing.T) {
+	f := newCeilingFixture(t)
+	_, ids := f.plan(fptr(10), fptr(0), fptr(100), 5, 5)
+	f.setBasis(ids[0], "", "")
+	f.setBasis(ids[1], "", "")
+	require.NoError(t, f.amend(ids[0], 4))
+	require.ErrorIs(t, f.amend(ids[0], 6), ErrLadderAmendInvalid, "assumed on-demand: 6+5 > 10")
+	require.NoError(t, f.amend(ids[0], 5))
+}
+
+func TestLadderAmendRunTotal_SingleCommitmentBasisRun(t *testing.T) {
+	f := newCeilingFixture(t)
+	runID, ids := f.plan(fptr(100), fptr(0), fptr(100), 4, 4)
+	for _, id := range ids {
+		f.setBasis(id, "commitment", "1/2")
+	}
+	_, err := f.store.db.Exec(f.ctx, "UPDATE ladder_runs SET total_hourly_commit=8 WHERE id=$1", runID)
+	require.NoError(t, err)
+	require.ErrorIs(t, f.amend(ids[0], 5), ErrLadderAmendInvalid, "run total 9 > original 8")
+	require.NoError(t, f.amend(ids[0], 3))
+	require.NoError(t, f.amend(ids[1], 5), "run total back to 8")
+}
+
 func seedNettingRows(t *testing.T, f *ceilingFixture, basis LadderAmountBasis, rate string, amounts ...string) {
 	t.Helper()
 	for _, a := range amounts {

known_issues/05_config_store_postgres.md

@@ -128,3 +128,10 @@
 **Resolved by:** `internal/api/db_rate_limiter.go::Allow` no longer issues a `SELECT ... FOR UPDATE` followed by a separate INSERT/UPDATE. Replaced with a single atomic `INSERT ... ON CONFLICT (id) DO UPDATE` that uses `CASE WHEN rate_limits.reset_time < $now THEN 1 ELSE rate_limits.count + 1 END` to handle both the fresh-insert / window-expired-reset / increment branches in one statement, with `RETURNING count, reset_time` driving the allow/deny decision. The dead `rateLimitOutcome` and `execAndCommit` helpers were removed. Three new integration tests in `internal/api/db_rate_limiter_integration_test.go` (build tag `integration`): `TestDBRateLimiter_ConcurrentFirstRequest` (20 goroutines hit the same id, all allowed, no SQLSTATE 23505), `TestDBRateLimiter_WindowExpiry_AtomicReset` (pre-seeded past reset_time → next call resets count to 1), `TestDBRateLimiter_ExceedsLimitDenies` (sanity: 3-allow + 2-deny burst with MaxAttempts=3). Documented trade-off: under sustained over-limit traffic, `count` may temporarily drift past MaxAttempts because the upsert always runs; the limiter still denies correctly and `cleanup()` evicts expired rows on its 24h cycle.
 
 **Implementation plan:** replace the SELECT+INSERT with a single atomic `INSERT ... ON CONFLICT (id) DO UPDATE SET count = CASE WHEN reset_time < $now THEN 1 ELSE count + 1 END, reset_time = CASE WHEN reset_time < $now THEN $newResetTime ELSE reset_time END, updated_at = $now RETURNING count, reset_time`. Drop the `queryErr/count/existingResetTime` arguments from `rateLimitOutcome`. Tests: `TestDBRateLimiter_ConcurrentFirstRequest` (20 goroutines hit same id, assert no SQLSTATE 23505 + final count = 20) and `TestDBRateLimiter_WindowExpiry_AtomicReset` (pre-seed past reset_time, single call resets count to 1).
+
+## MEDIUM: nothing stops a future ladder executor from claiming a legacy on-demand row
+
+**File**: `internal/config/store_postgres_ladder_lifecycle.go`, `ClaimDueLadderTranches`
+**Description**: Legacy `on_demand_equivalent` tranches net at rate 1 and the amend ceiling assumes NULL-basis rows are on-demand (#886 left them unknown). The claim query does not filter by basis, so an executor that buys `amount_usd_hr` as commitment dollars would overbuy a legacy row. No executor exists yet.
+**Impact**: None today; blocks executor enablement. Track on #946 and the executor issue.
+**Status:** Open