fix(lua): bound prewarm by queue depth, not by the ceiling

Production oscillated between ~46 MB and ~56 MB RSS with the openai pool
cycling 1 -> 10..12 -> 1 states every couple of minutes, while peak_in_use
never went above 2.

Cause: the batch prewarm sized itself purely on the adapter's ceiling. With
max_concurrent summing to 76, growStep is 8, so any two overlapping requests
warmed 8 states — 6 more than anything was waiting for. A minute later the
janitor correctly reclaimed the surplus, the next pair of overlapping requests
warmed 8 again, and the pool churned boot/discard forever. The elasticity was
working; the growth signal was simply wrong.

Prewarm is now bounded by BOTH limits: the ceiling still caps the step, but the
batch never exceeds p.waiting, the number of goroutines actually blocked on the
pool. Overlapping-but-not-queued traffic (the common case) creates exactly the
states it uses; a genuinely queued burst still ramps in one jump.

TestContentionBatchPrewarms is rewritten to queue real waiters instead of
relying on the ceiling to imply demand, and TestNoPrewarmWithoutWaiters pins the
production shape: two overlapping requests against a 76-wide adapter must create
exactly 2 states.

internal/lua/vm_test.go

@@ -1217,18 +1217,97 @@ func TestSequentialTrafficDoesNotBatch(t *testing.T) {
 	}
 }
 
-// TestContentionBatchPrewarms: when concurrent demand exceeds the warm set, the
-// pool warms a whole grow step instead of one state per request.
+// TestContentionBatchPrewarms: when callers are actually QUEUED behind a full
+// pool, the pool warms extra states for them instead of booting one per request.
+// The batch is bounded by the queue depth, not by the ceiling, so it can never
+// warm states nobody is waiting for.
 func TestContentionBatchPrewarms(t *testing.T) {
 	vm := NewVM(freshAdapterDir(t))
 	if err := vm.Start(); err != nil {
 		t.Fatalf("start: %v", err)
 	}
 	defer vm.Stop()
-	vm.ConfigureConcurrency(map[string]int{"openai": 32}) // grow step = 4
+	// ceiling 32 -> step cap 4; queue depth will be the real bound
+	vm.ConfigureConcurrency(map[string]int{"openai": 32})
 	p := poolOf(t, vm, "openai")
 
-	// hold the first state, then miss while it is busy -> contended
+	// Fill the pool to its ceiling-independent state: hold 1, then queue 3 more
+	// so waiting == 3 when the next miss happens.
+	p.mu.Lock()
+	p.maxW = 1 // force the next acquires to queue
+	p.mu.Unlock()
+	w1, err := p.acquire()
+	if err != nil {
+		t.Fatalf("acquire 1: %v", err)
+	}
+	done := make(chan *worker, 3)
+	for i := 0; i < 3; i++ {
+		go func() {
+			w, err := p.acquire()
+			if err != nil {
+				done <- nil
+				return
+			}
+			done <- w
+		}()
+	}
+	// wait until all three are queued
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		p.mu.Lock()
+		w := p.waiting
+		p.mu.Unlock()
+		if w >= 3 {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("waiters never queued, waiting=%d", w)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	// raise the ceiling: the next miss sees waiting=3 and prewarms a batch
+	p.setMax(32)
+	got := make([]*worker, 0, 3)
+	for i := 0; i < 3; i++ {
+		select {
+		case w := <-done:
+			if w == nil {
+				t.Fatal("a queued acquire failed")
+			}
+			got = append(got, w)
+		case <-time.After(5 * time.Second):
+			t.Fatal("queued acquires never completed after the ceiling was raised")
+		}
+	}
+	created, _, _ := poolCounts(p)
+	if created < 4 {
+		t.Fatalf("queued demand must be satisfied, created=%d want >=4", created)
+	}
+	if created > 32 {
+		t.Fatalf("prewarm exceeded the ceiling, created=%d", created)
+	}
+	p.release(w1)
+	for _, w := range got {
+		p.release(w)
+	}
+}
+
+// TestNoPrewarmWithoutWaiters is the production regression this bound fixes: two
+// concurrent requests against a wide adapter (ceiling 76, step 8) must NOT warm
+// eight states. Warming for nobody meant the janitor discarded the surplus a
+// minute later, so RSS oscillated between ~46 MB and ~56 MB forever.
+func TestNoPrewarmWithoutWaiters(t *testing.T) {
+	vm := NewVM(freshAdapterDir(t))
+	if err := vm.Start(); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	defer vm.Stop()
+	vm.ConfigureConcurrency(map[string]int{"openai": 76})
+	p := poolOf(t, vm, "openai")
+
+	// two overlapping requests: the second misses while the first holds its
+	// state, but nobody is blocked, so no batch may be warmed
 	w1, err := p.acquire()
 	if err != nil {
 		t.Fatalf("acquire 1: %v", err)
@@ -1237,21 +1316,10 @@ func TestContentionBatchPrewarms(t *testing.T) {
 	if err != nil {
 		t.Fatalf("acquire 2: %v", err)
 	}
-	// the batch is booted in the background; wait for it to land
-	deadline := time.Now().Add(5 * time.Second)
-	for {
-		created, _, _ := poolCounts(p)
-		if created >= 5 { // 2 checked out + 3 prewarmed
-			break
-		}
-		if time.Now().After(deadline) {
-			t.Fatalf("contended miss did not batch prewarm, created=%d", created)
-		}
-		time.Sleep(5 * time.Millisecond)
-	}
+	time.Sleep(100 * time.Millisecond) // give any background prewarm a chance
 	created, _, _ := poolCounts(p)
-	if created > 32 {
-		t.Fatalf("prewarm exceeded the ceiling, created=%d", created)
+	if created != 2 {
+		t.Fatalf("two overlapping requests must create exactly 2 states, got %d", created)
 	}
 	p.release(w1)
 	p.release(w2)