a request could ask for a warm floor but not for a ceiling (v0.4.33)

/v1/runpod/scale set a FLOOR. "Give me 8 pods" meant "hold at least 8
ready", and the autoscaler stayed free to grow the pool to the configured
max_pods for as long as work kept arriving. A client running a staged
batch -- 8 pods for the first stage, 32 for the next -- had no way to
bound a stage at all, and nothing in the reply said so.

Observed 2026-10-18: a run that asked for 8 pods was at 13 and climbing
toward 128. Every pod past the plan is ~700s of paid boot producing
nothing, which is why the realised cost per page came out 16.5x the
measured baseline while every component behaved exactly as written.

A lease may now also carry `max_pods`, under the same rule `gpu_types`
already follows there: it may only ever NARROW the operator's ceiling, a
larger number is clamped and reported in `notes`, and it dies with the
lease. `pods: 0` with a `max_pods` is a cap with no floor -- what a
client with a long tail of work wants; `pods: 0` alone releases both.

The ceiling has to be read by EVERY path that decides whether to rent,
or it is a race between the grower that honours it and the one that does
not: _pods_for_demand, _grow_reason, _pods_needed and the warm-floor
top-up all now read effective_max_pods(). A leased floor is clamped to
it too, since a floor above a ceiling is a fleet that oscillates for as
long as the lease lasts. A configured max_pods of 0 still means zero --
that is the operator having turned renting off, not a lease.

22 tests, including a guard that no growth path reads mcfg.max_pods
directly, so a path added later cannot silently escape every cap.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.4.32"
+__version__ = "0.4.33"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/app.py

@@ -962,6 +962,15 @@ class RunpodScaleRequest(BaseModel):
     pods: int = Field(..., ge=0,
                       description="Warm pods to hold, clamped to the model's max_pods. "
                                   "0 releases the lease.")
+    max_pods: Optional[int] = Field(
+        None, ge=0,
+        description="Cap the fleet at this many pods for the life of this "
+                    "lease. May only NARROW the model's configured max_pods: a "
+                    "larger number is clamped to it and reported back in "
+                    "`notes`. `pods` is a FLOOR and says nothing about how "
+                    "large the pool may grow, so a staged batch that wants "
+                    "8 pods for its first stage sets both. 0 drops this "
+                    "lease's own cap; omit to leave it as it is.")
     teardown: bool = Field(False,
                            description="Give the surplus pods back NOW rather than "
                                        "waiting out scale_down_after_s. Never goes "
@@ -1034,8 +1043,19 @@ async def runpod_scale(body: RunpodScaleRequest, request: Request):
 
     Deliberately NOT admin-scoped: the gate is per model
     (``allow_client_scale``), the number is clamped to ``max_pods``, and the
-    lease expires (``client_scale_ttl_s``, default 30 min) so a client that dies
-    holding four A40s stops paying for them by itself. The account's $/hr and
+    ``max_pods`` is the other half of that, and the half that bounds what a
+    batch may cost. ``pods`` is a FLOOR: it says what to hold ready, and the
+    autoscaler is still free to grow the pool to the configured ``max_pods``
+    for as long as work keeps arriving. A client running a staged batch — 8
+    pods for the first stage, 32 for the next — needs to say how large the
+    fleet may BECOME, which it does by sending both. Like ``gpu_types`` it may
+    only ever narrow the operator's own number and it dies with the lease.
+    ``pods: 0`` with a ``max_pods`` keeps the cap and holds nothing warm, which
+    is what a client with a long tail of work wants; ``pods: 0`` alone releases
+    the whole lease.
+
+    The lease expires (``client_scale_ttl_s``, default 30 min) so a client that
+    dies holding four A40s stops paying for them by itself. The account's $/hr and
     spend caps still refuse to provision regardless of what is asked here.
     """
     from codai.api import runpod_worker
@@ -1051,7 +1071,8 @@ async def runpod_scale(body: RunpodScaleRequest, request: Request):
     task = asyncio.create_task(asyncio.to_thread(
         runpod_worker.set_client_pods,
         body.model, body.pods, teardown=bool(body.teardown),
-        gpu_types=body.gpu_types, allow_other_gpus=body.allow_other_gpus))
+        gpu_types=body.gpu_types, allow_other_gpus=body.allow_other_gpus,
+        max_pods=body.max_pods))
 
     def _done(t):
         try:

codai/api/runpod_worker.py

@@ -2970,6 +2970,24 @@ class RunpodPodPool:
         self._client_floor = 0
         self._client_floor_until = 0.0
         self._client_floor_logged = -1
+        #: A CEILING the same lease may set, and when it lapses. 0/0 = nobody
+        #: asked, and the configured `max_pods` is the only ceiling.
+        #:
+        #: The floor and the ceiling are not the same request. A client that
+        #: says "hold 8 warm" has said nothing about how large the fleet may
+        #: become, and the autoscaler is free to grow it to `max_pods` as long
+        #: as work keeps arriving -- which is exactly what happened on
+        #: 2026-10-18: a staged run asked for 8 pods, got 13, and was heading
+        #: for 128, because the only number a request could set was the floor.
+        #: Each pod beyond the plan added ~700 s of paid boot producing nothing,
+        #: and the realised cost per page came out 16.5x the measured baseline.
+        #:
+        #: So a lease may ALSO narrow the ceiling, and like `gpu_types` it may
+        #: only ever narrow it: the operator's `max_pods` remains the budget,
+        #: and there is deliberately no way for a request to raise it.
+        self._client_cap = 0
+        self._client_cap_until = 0.0
+        self._client_cap_logged = -1
         #: GPU constraints a client attached to its lease. They expire with it.
         #: Only ever NARROWER than the operator's own settings -- see
         #: `narrow_gpu_choice`.
@@ -4165,7 +4183,7 @@ class RunpodPodPool:
         if mean <= 0:
             mean = float(threshold)
         want = int(-(-load // mean))          # ceil, in weighted pod-equivalents
-        return max(1, min(int(self.mcfg.max_pods), want))
+        return max(1, min(self.effective_max_pods(), want))
 
     def _boot_cancelled(self, epoch: int) -> bool:
         """Was this boot released while it was still booting?
@@ -4209,7 +4227,7 @@ class RunpodPodPool:
         that start billing together.
         """
         in_hand = len(self.pods) + self._booting
-        if in_hand >= self.mcfg.max_pods:
+        if in_hand >= self.effective_max_pods():
             return ""
         parallel = max(1, int(getattr(self.mcfg, "scale_up_max_parallel_boots", 1) or 1))
         if self._booting >= parallel:
@@ -4402,7 +4420,7 @@ class RunpodPodPool:
                 self._note_denied()
                 raise NoPodRightNow(
                     f"no pod slot for {self.model_key!r} right now "
-                    f"(max_pods={self.mcfg.max_pods})")
+                    f"(max_pods={self.effective_max_pods()})")
 
             # At max_pods with every pod at its in-flight ceiling, or someone else
             # is provisioning: wait for a slot rather than piling on. Counted
@@ -4414,7 +4432,7 @@ class RunpodPodPool:
                 if remaining <= 0:
                     raise RunpodError(
                         f"RunPod: no pod slot available for {self.model_key!r} "
-                        f"(max_pods={self.mcfg.max_pods}, "
+                        f"(max_pods={self.effective_max_pods()}, "
                         f"max_inflight_per_pod={self.mcfg.max_inflight_per_pod}) "
                         f"within {timeout}s.")
                 self._waiting += 1
@@ -4582,7 +4600,58 @@ class RunpodPodPool:
             return 0
         if time.time() >= self._client_floor_until:
             return 0
-        return min(self._client_floor, max(1, self.mcfg.max_pods))
+        # Against the EFFECTIVE ceiling, so a lease that later narrowed its own
+        # max_pods does not keep a floor above it -- a floor the pool would
+        # have to grow to and a ceiling forbidding it is a fleet that oscillates
+        # for as long as the lease lasts.
+        return min(self._client_floor, self.effective_max_pods())
+
+    def client_cap(self) -> int:
+        """The pod CEILING a client leased, 0 once the lease has run out.
+
+        Read-only and cheap, exactly like `client_floor`, so the scaler, the
+        status page and the endpoint all see the same number within a pass.
+
+        getattr, for the same reason `_slots_of` initialises on demand: not
+        every pool object this is asked about went through `__init__`, and a
+        READ-ONLY question about the ceiling must never be the thing that
+        raises on the path that decides whether to rent a GPU. No lease is the
+        normal answer anyway.
+        """
+        leased = int(getattr(self, "_client_cap", 0) or 0)
+        if leased <= 0:
+            return 0
+        if time.time() >= float(getattr(self, "_client_cap_until", 0.0) or 0.0):
+            return 0
+        return min(leased, max(1, self.mcfg.max_pods))
+
+    def effective_max_pods(self) -> int:
+        """The ceiling on fleet size that applies right now.
+
+        The configured `max_pods`, narrowed by a client lease if one is in
+        force. Every path that decides whether to rent another pod reads THIS
+        rather than `mcfg.max_pods`, because a ceiling only half of them honour
+        is not a ceiling -- it is a race between the grower that respects it and
+        the one that does not.
+
+        A LEASE can never drive this to zero: "do not grow past N" is not a
+        request to stop serving, and a lease asking for 0 is read as releasing
+        its cap. A configured `max_pods` of 0 is different -- that is the
+        operator having turned pod renting off for this model, and it stays 0.
+        """
+        configured = int(self.mcfg.max_pods or 0)
+        capped = self.client_cap()
+        if capped <= 0 or configured <= 0:
+            return configured
+        out = max(1, min(configured, capped))
+        if out != getattr(self, "_client_cap_logged", -1):
+            print(f"[runpod] {self.model_key!r}: fleet ceiling {out} from a "
+                  f"client lease (configured max_pods {configured})", flush=True)
+            try:
+                self._client_cap_logged = out
+            except Exception:
+                pass
+        return out
 
     def client_gpu_narrow(self) -> dict:
         """The GPU narrowing in force right now, or {} once the lease lapses.
@@ -4638,14 +4707,36 @@ class RunpodPodPool:
         return out
 
     def set_client_floor(self, pods: int, gpu_types=None,
-                         allow_other_gpus=None) -> dict:
-        """Lease (or release, with 0) a client-requested warm floor.
+                         allow_other_gpus=None, max_pods=None) -> dict:
+        """Lease (or release, with 0) a client-requested warm floor and ceiling.
 
         Clamped to max_pods: the configured ceiling is the budget, and a client
         asking for more gets the ceiling rather than an error — it asked for
         "as many as you can", and failing the call would leave it with none.
+
+        ``max_pods`` narrows the CEILING for the life of the lease. It is a
+        separate question from the floor and it needs its own answer: a client
+        that knows it is running a staged batch -- 8 pods for the first stage,
+        32 for the next -- can bound what the stage may cost, where asking for
+        8 warm pods bounds nothing at all. Only ever narrower than the
+        operator's configured `max_pods`; ``0`` drops the lease's own cap and
+        ``None`` leaves it as it was.
+
+        A floor above the ceiling is not refused: the ceiling wins, because it
+        is the number that bounds spending and the floor is a convenience. The
+        reply reports what was granted either way, so a caller that asked for
+        something incoherent can see that it did.
         """
-        cap = max(1, int(self.mcfg.max_pods))
+        configured = max(1, int(self.mcfg.max_pods))
+        # The lease's own ceiling first: the floor is clamped to it below, so a
+        # caller sending {"pods": 32, "max_pods": 8} gets 8 and not 32 warm
+        # pods under an 8-pod cap, which is a fleet that can never settle.
+        if max_pods is None:
+            leased_cap = self.client_cap()
+        else:
+            leased_cap = max(0, int(max_pods))
+            leased_cap = min(leased_cap, configured) if leased_cap else 0
+        cap = min(configured, leased_cap) if leased_cap else configured
         want = max(0, int(pods))
         granted = min(want, cap)
         if granted:
@@ -4653,6 +4744,13 @@ class RunpodPodPool:
             # its mind, and retiring what it just asked for would be absurd.
             self._release_pending = False
         ttl = max(60, int(getattr(self.mcfg, "client_scale_ttl_s", 1800) or 1800))
+        # A cap with no floor is a legitimate lease: "grow no further than 8,
+        # but hold nothing warm for me" is what a client with a long tail of
+        # work wants, and releasing the cap because the floor went to 0 would
+        # take the ceiling away at exactly the moment it still matters. So the
+        # cap lapses with its OWN deadline, and `pods: 0` clears it only when
+        # the same call did not ask for one.
+        keep_cap = leased_cap if (max_pods is not None or granted) else 0
         # Raises ValueError when the caller narrowed to nothing it is allowed to
         # have; the endpoint turns that into a 400 that names the allowed set.
         narrow = narrow_gpu_choice(self.mcfg, gpu_types, allow_other_gpus)
@@ -4660,6 +4758,9 @@ class RunpodPodPool:
             self._client_floor = granted
             self._client_floor_until = (time.time() + ttl) if granted else 0.0
             self._client_floor_logged = -1
+            self._client_cap = keep_cap
+            self._client_cap_until = (time.time() + ttl) if keep_cap else 0.0
+            self._client_cap_logged = -1
             if granted:
                 self._client_gpu_types = (list(narrow["gpu_types"])
                                           if gpu_types else [])
@@ -4673,14 +4774,31 @@ class RunpodPodPool:
                 self._client_allow_other = None
         print(f"[runpod] {self.model_key!r}: client asked for {want} warm pod(s), "
               f"granted {granted} (max_pods {cap}) for {ttl}s", flush=True)
+        if max_pods is not None:
+            print(f"[runpod] {self.model_key!r}: this lease caps the fleet at "
+                  f"{keep_cap or configured} pod(s) "
+                  f"(configured max_pods {configured}"
+                  + ("; the lease asked for more and was narrowed to it"
+                     if int(max_pods or 0) > configured else "")
+                  + (")" if keep_cap else "; the lease's own cap is released)"),
+                  flush=True)
         if granted and (gpu_types or allow_other_gpus is not None):
             print(f"[runpod] {self.model_key!r}: this lease asks for "
                   f"{narrow['gpu_types'] or 'any card'}"
                   f"{'' if narrow['allow_other_gpus'] else ' and nothing else'}"
                   + ("".join(f"; {n}" for n in narrow["notes"])), flush=True)
         out = {"model": self.model_key, "requested": want, "granted": granted,
-               "max_pods": cap, "expires_in_s": ttl if granted else 0,
+               "max_pods": cap, "configured_max_pods": configured,
+               "client_max_pods": keep_cap,
+               "expires_in_s": ttl if (granted or keep_cap) else 0,
                "clamped": want > cap}
+        if max_pods is not None and int(max_pods or 0) > configured:
+            # Reported for the same reason a narrowed gpu_types is: a caller
+            # that believes it capped the fleet at 256 would read the bill as
+            # coming from a cap it never had.
+            out.setdefault("notes", []).append(
+                f"max_pods {int(max_pods)} narrowed to the configured "
+                f"ceiling {configured}")
         if granted and (gpu_types or allow_other_gpus is not None):
             # Reported back, because a constraint that was silently dropped is
             # worse than one refused: the caller would read the numbers as
@@ -4688,7 +4806,10 @@ class RunpodPodPool:
             out["gpu_types"] = narrow["gpu_types"]
             out["allow_other_gpus"] = narrow["allow_other_gpus"]
             if narrow["notes"]:
-                out["notes"] = narrow["notes"]
+                # Extend: a narrowed max_pods may already have left a note
+                # here, and replacing the list would hide whichever of the two
+                # constraints was reported second.
+                out.setdefault("notes", []).extend(narrow["notes"])
         return out
 
     def effective_min_pods(self) -> int:
@@ -4714,7 +4835,7 @@ class RunpodPodPool:
             print(f"[runpod] {self.model_key!r}: warm floor {leased} from a client "
                   f"lease (configured {scheduled})", flush=True)
             self._client_floor_logged = leased
-        return max(scheduled, min(leased, max(1, self.mcfg.max_pods)))
+        return max(scheduled, min(leased, self.effective_max_pods()))
 
     def _pods_needed(self, serving: list) -> int:
         """How many pods the CURRENT concurrency justifies.
@@ -4727,7 +4848,8 @@ class RunpodPodPool:
         threshold = max(1, int(self.mcfg.scale_up_inflight_per_pod or 1))
         total = sum(p.inflight for p in serving)
         need = -(-total // threshold)          # ceil, in integers
-        return max(self.effective_min_pods(), min(int(self.mcfg.max_pods), need))
+        return max(self.effective_min_pods(),
+                   min(self.effective_max_pods(), need))
 
     def _scale_down_victim(self, serving: list):
         """Which surplus pod to give back.
@@ -5092,7 +5214,7 @@ class RunpodPodPool:
                                               "scale_up_max_parallel_boots", 4) or 4))
                 want = min(keep_min - healthy_in_hand,
                            max(0, parallel - self._booting),
-                           max(0, int(self.mcfg.max_pods or 0) - billing_in_hand))
+                           max(0, self.effective_max_pods() - billing_in_hand))
                 if want <= 0:
                     return
                 self._booting += want
@@ -5657,6 +5779,10 @@ def pools_schedule_status() -> list:
             "min_pods": cfg.min_pods,
             "effective_min_pods": pool.effective_min_pods(),
             "max_pods": cfg.max_pods,
+            # The ceiling actually in force, and where it came from: a fleet
+            # that stopped growing short of max_pods must say why.
+            "effective_max_pods": pool.effective_max_pods(),
+            "client_max_pods": pool.client_cap(),
             "healthy_pods": healthy,
             "data_center": cfg.data_center or "",
             # The client lease, so a bill nobody expected has a visible cause.
@@ -5745,7 +5871,8 @@ def get_live_pool(key: str):
 
 
 def _scale_live_pool(pool, key: str, pods: int, teardown: bool = False,
-                     gpu_types=None, allow_other_gpus=None) -> dict:
+                     gpu_types=None, allow_other_gpus=None,
+                     max_pods=None) -> dict:
     """`set_client_pods` for a pool addressed directly rather than by model."""
     mcfg = getattr(pool, "mcfg", None)
     if mcfg is not None and not getattr(mcfg, "allow_client_scale", False):
@@ -5754,7 +5881,8 @@ def _scale_live_pool(pool, key: str, pods: int, teardown: bool = False,
             '"allow_client_scale": true on the runpod block of a model that '
             "uses it")
     out = pool.set_client_floor(pods, gpu_types=gpu_types,
-                                allow_other_gpus=allow_other_gpus)
+                                allow_other_gpus=allow_other_gpus,
+                                max_pods=max_pods)
     if teardown:
         out["teardown"] = pool.shed_now()
     out["pool"] = out.get("model") or key
@@ -5882,7 +6010,8 @@ def release_pod(pod_id: str, reason: str = "released by client") -> dict:
 
 
 def set_client_pods(model: str, pods: int, teardown: bool = False,
-                    gpu_types=None, allow_other_gpus=None) -> dict:
+                    gpu_types=None, allow_other_gpus=None,
+                    max_pods=None) -> dict:
     """Lease a warm floor for one model on a client's behalf.
 
     Raises LookupError when the model has no RunPod block, and PermissionError
@@ -5902,6 +6031,12 @@ def set_client_pods(model: str, pods: int, teardown: bool = False,
     fallback's price for a result it cannot use. Raises ValueError when the
     narrowing leaves nothing allowed, so the caller hears about a mistake
     instead of silently getting the operator's default.
+
+    ``max_pods`` caps the fleet for the life of the lease, and may only narrow
+    the operator's own ceiling. A warm floor says what to hold ready; it says
+    nothing about how large the pool may become, so without this a client
+    running a staged batch could ask for its first stage and still be grown to
+    the configured maximum by the autoscaler.
     """
     key = str(model or "").strip()
     if not key:
@@ -5916,7 +6051,8 @@ def set_client_pods(model: str, pods: int, teardown: bool = False,
     if existing is not None:
         return _scale_live_pool(existing, key, pods, teardown=teardown,
                                 gpu_types=gpu_types,
-                                allow_other_gpus=allow_other_gpus)
+                                allow_other_gpus=allow_other_gpus,
+                                max_pods=max_pods)
     block = model_runpod_block(key)
     if not block:
         raise LookupError(
@@ -5936,7 +6072,8 @@ def set_client_pods(model: str, pods: int, teardown: bool = False,
     # has called yet is exactly when it is worth pre-warming.
     pool = get_model_pod_pool(key, _model_entry(key), block)
     out = pool.set_client_floor(pods, gpu_types=gpu_types,
-                                allow_other_gpus=allow_other_gpus)
+                                allow_other_gpus=allow_other_gpus,
+                                max_pods=max_pods)
     if teardown:
         # After the floor is lowered, so the shed measures against the new one.
         out["teardown"] = pool.shed_now()

tests/test_runpod_client_pod_ceiling.py

+"""A request may say how large the fleet is allowed to BECOME, not only how
+many pods to hold warm.
+
+/v1/runpod/scale set a FLOOR and nothing else. "Give me 8 pods" meant "hold at
+least 8 ready", and the autoscaler stayed free to grow the pool to the
+configured max_pods for as long as work kept arriving. A client running a
+staged batch -- 8 pods for the first stage, 32 for the next -- had no way to
+bound a stage at all.
+
+Observed 2026-10-18: a run that asked for 8 pods was at 13 and climbing toward
+128. Each pod past the plan is ~700 s of paid boot producing nothing, which is
+why the realised cost per page came out 16.5x the measured baseline while every
+pod was behaving correctly. Nothing was broken; the number the client sent was
+simply not the number it needed to send.
+
+So a lease may also carry `max_pods`, under the same rule `gpu_types` follows
+there: it may only ever NARROW the operator's ceiling, and it dies with the
+lease. The properties these tests pin, in order of what getting them wrong
+would cost:
+
+  1. a cap is never WIDENED -- a request must not be able to raise what the
+     operator budgeted, which is the whole reason this is not admin-scoped
+  2. EVERY growth path honours it, because a ceiling half of them respect is a
+     race, not a ceiling
+  3. it expires with the lease, so a client that dies holding a cap does not
+     leave the pool permanently narrowed
+  4. a floor and a cap are separate questions, and asking one must not silently
+     answer the other
+"""
+
+import time
+import types
+
+import pytest
+
+from codai.api import runpod_worker as rw
+
+
+def _pool(max_pods=128, min_pods=0, threshold=16, parallel=4, ttl=1800):
+    class _Acct:
+        enabled = True
+
+    return rw.RunpodPodPool("m", _Acct(), rw.parse_model_runpod({
+        "max_pods": max_pods,
+        "min_pods": min_pods,
+        "scale_up_inflight_per_pod": threshold,
+        "scale_up_max_parallel_boots": parallel,
+        "allow_client_scale": True,
+        "client_scale_ttl_s": ttl,
+    }), "m")
+
+
+def _pod(inflight=0, healthy=True, draining=False):
+    return types.SimpleNamespace(inflight=inflight, healthy=healthy,
+                                 draining=draining, pod_id="p", url="u",
+                                 last_used=0.0, gpu="NVIDIA A40")
+
+
+# --- 1. a request may only ever narrow ------------------------------------
+
+def test_a_lease_cannot_raise_the_configured_ceiling():
+    """The reason this endpoint can stay unauthenticated."""
+    p = _pool(max_pods=8)
+    out = p.set_client_floor(2, max_pods=1024)
+    assert p.effective_max_pods() == 8
+    assert out["configured_max_pods"] == 8
+    assert out["client_max_pods"] == 8
+
+
+def test_a_narrowed_ceiling_that_was_clamped_says_so():
+    """A cap silently widened would be read as a cap that held."""
+    p = _pool(max_pods=8)
+    out = p.set_client_floor(2, max_pods=1024)
+    assert any("narrowed" in n for n in out.get("notes", [])), out
+
+
+def test_a_narrowing_lease_is_obeyed():
+    p = _pool(max_pods=128)
+    p.set_client_floor(8, max_pods=8)
+    assert p.effective_max_pods() == 8
+
+
+def test_no_lease_leaves_the_operator_ceiling_alone():
+    p = _pool(max_pods=128)
+    assert p.effective_max_pods() == 128
+    assert p.client_cap() == 0
+
+
+def test_an_operator_who_configured_zero_pods_still_gets_zero():
+    """A LEASE can never drive the ceiling to zero, but max_pods: 0 is the
+    operator having turned pod renting off, and that is not a lease."""
+    p = _pool(max_pods=1)
+    # `parse_model_runpod` floors the configured value at 1, so 0 only reaches
+    # the pool by direct assignment -- which is how the scale-out fixtures
+    # express "this pool may not rent", and the growth paths must keep
+    # answering it the same way.
+    p.mcfg.max_pods = 0
+    assert p.effective_max_pods() == 0
+    p.set_client_floor(4, max_pods=4)
+    assert p.effective_max_pods() == 0
+
+
+def test_a_cap_is_never_zero():
+    """'Do not grow past N' is not a request to stop serving."""
+    p = _pool(max_pods=128)
+    p.set_client_floor(0, max_pods=0)
+    assert p.effective_max_pods() == 128
+    p.set_client_floor(1, max_pods=1)
+    assert p.effective_max_pods() == 1
+
+
+# --- 2. every growth path honours it --------------------------------------
+
+def test_the_grower_stops_at_the_leased_ceiling():
+    """_grow_reason is the autoscaler's own question."""
+    p = _pool(max_pods=128, threshold=16)
+    p.set_client_floor(0, max_pods=2)
+    busy = [_pod(inflight=16), _pod(inflight=16)]
+    p.pods = list(busy)
+    assert p._grow_reason(busy) == "", "a saturated fleet still must not grow past the cap"
+
+
+def test_the_grower_would_have_grown_without_the_cap():
+    """Guards the test above: the cap is what stopped it, not the state."""
+    p = _pool(max_pods=128, threshold=16)
+    busy = [_pod(inflight=16), _pod(inflight=16)]
+    p.pods = list(busy)
+    assert p._grow_reason(busy), "the trigger must really fire, or the cap proves nothing"
+
+
+def test_demand_is_bounded_by_the_leased_ceiling():
+    """_pods_for_demand: what a measured queue justifies renting."""
+    p = _pool(max_pods=128, threshold=16)
+    p.set_client_floor(0, max_pods=4)
+    serving = [_pod(inflight=16)]
+    p.pods = list(serving)
+    p._denied = [time.time()] * 2048          # a very deep queue
+    assert p._pods_for_demand(serving) <= 4
+
+
+def test_scale_down_targets_the_leased_ceiling():
+    """_pods_needed decides how many to KEEP, so the cap must shed too."""
+    p = _pool(max_pods=128, threshold=1)
+    p.set_client_floor(0, max_pods=3)
+    serving = [_pod(inflight=8) for _ in range(8)]
+    assert p._pods_needed(serving) == 3
+
+
+def test_the_warm_floor_top_up_cannot_exceed_the_leased_ceiling():
+    """The floor is a convenience; the ceiling is the budget."""
+    p = _pool(max_pods=128, min_pods=16)
+    p.set_client_floor(0, max_pods=2)
+    # The operator's own min_pods is higher than the cap. The cap wins, because
+    # the alternative is renting cards the client explicitly refused.
+    assert p.effective_max_pods() == 2
+    assert p.effective_min_pods() <= 16
+    healthy_in_hand = 0
+    billing_in_hand = 0
+    want = min(p.effective_min_pods() - healthy_in_hand, 4,
+               max(0, p.effective_max_pods() - billing_in_hand))
+    assert want <= 2
+
+
+# --- 3. the cap dies with the lease ---------------------------------------
+
+def test_the_cap_expires():
+    p = _pool(max_pods=128, ttl=60)
+    p.set_client_floor(2, max_pods=2)
+    assert p.effective_max_pods() == 2
+    p._client_cap_until = time.time() - 1
+    assert p.client_cap() == 0
+    assert p.effective_max_pods() == 128
+
+
+def test_releasing_the_whole_lease_releases_the_cap():
+    """`pods: 0` alone is a client going away; it keeps nothing."""
+    p = _pool(max_pods=128)
+    p.set_client_floor(4, max_pods=4)
+    p.set_client_floor(0)
+    assert p.client_cap() == 0
+    assert p.client_floor() == 0
+    assert p.effective_max_pods() == 128
+
+
+def test_a_cap_survives_a_floor_of_zero_when_the_same_call_asks_for_it():
+    """'Grow no further than 8, but hold nothing warm' is a real request: a
+    client with a long tail of work wants the ceiling and not the floor."""
+    p = _pool(max_pods=128)
+    p.set_client_floor(0, max_pods=8)
+    assert p.client_floor() == 0
+    assert p.client_cap() == 8
+    assert p.effective_max_pods() == 8
+
+
+def test_renewing_the_floor_alone_keeps_the_cap():
+    """max_pods omitted means 'leave it as it is', not 'drop it'."""
+    p = _pool(max_pods=128)
+    p.set_client_floor(4, max_pods=8)
+    p.set_client_floor(6)
+    assert p.client_cap() == 8
+    assert p.client_floor() == 6
+
+
+# --- 4. a floor and a cap are separate questions --------------------------
+
+def test_asking_for_warm_pods_alone_does_not_cap_the_fleet():
+    """The behaviour BEFORE this existed, and it must not change: a floor says
+    nothing about the ceiling, so an install that never sends max_pods keeps
+    the autoscaler it had."""
+    p = _pool(max_pods=128)
+    p.set_client_floor(8)
+    assert p.effective_max_pods() == 128
+
+
+def test_a_floor_above_the_cap_is_granted_as_the_cap():
+    """Incoherent, but refusing would leave the client with nothing. The
+    ceiling wins: it is the number that bounds the bill."""
+    p = _pool(max_pods=128)
+    out = p.set_client_floor(32, max_pods=8)
+    assert out["granted"] == 8
+    assert p.client_floor() == 8
+    assert p.effective_max_pods() == 8
+
+
+def test_lowering_a_cap_under_a_live_floor_lowers_the_floor_too():
+    """Otherwise the pool must grow to a floor a ceiling forbids, and
+    oscillates for as long as the lease lasts."""
+    p = _pool(max_pods=128)
+    p.set_client_floor(16)
+    assert p.client_floor() == 16
+    p.set_client_floor(16, max_pods=4)
+    assert p.effective_max_pods() == 4
+    assert p.client_floor() == 4
+
+
+# --- the endpoint carries it ----------------------------------------------
+
+def test_the_scale_endpoint_accepts_and_forwards_max_pods():
+    src = open("codai/api/app.py").read()
+    assert "max_pods: Optional[int] = Field(" in src
+    assert "max_pods=body.max_pods" in src
+
+
+def test_set_client_pods_forwards_it_for_both_ways_of_naming_a_pool():
+    """A model id and a live `pool:` key take different branches, and a cap
+    honoured on only one of them is a cap that depends on how it was asked."""
+    import inspect
+    for fn in (rw.set_client_pods, rw._scale_live_pool):
+        assert "max_pods" in inspect.signature(fn).parameters, fn.__name__
+    src = inspect.getsource(rw.set_client_pods)
+    assert src.count("max_pods=max_pods") >= 2, (
+        "both the live-pool branch and the by-model branch must pass it")
+
+
+def test_the_status_surface_reports_the_ceiling_in_force():
+    """A fleet that stopped growing short of max_pods must say why."""
+    src = inspect_source = open("codai/api/runpod_worker.py").read()
+    assert '"effective_max_pods": pool.effective_max_pods()' in src
+    assert '"client_max_pods": pool.client_cap()' in src
+
+
+def test_no_growth_path_still_reads_the_raw_configured_ceiling():
+    """The regression guard. `mcfg.max_pods` is the BUDGET; the growth paths
+    must read `effective_max_pods()`, and a new one added later that reaches
+    for the raw number would silently escape every client cap."""
+    import inspect
+    for name in ("_pods_for_demand", "_grow_reason", "_pods_needed"):
+        src = inspect.getsource(getattr(rw.RunpodPodPool, name))
+        assert "mcfg.max_pods" not in src, f"{name} reads the raw ceiling"
+    src = inspect.getsource(rw.RunpodPodPool.maintain)
+    assert "self.mcfg.max_pods or 0" not in src, "the warm-floor top-up reads the raw ceiling"

tests/test_runpod_client_scale.py

@@ -377,7 +377,9 @@ def test_the_demand_path_already_had_the_guard_and_keeps_it():
     # meant the fleet could grow by one pod per boot however much was queued.
     # The guard is still here, it just reads how many rather than whether.
     assert "self._booting >= parallel" in blk
-    assert "in_hand >= self.mcfg.max_pods" in blk
+    # Reads the EFFECTIVE ceiling since v0.4.33: the configured max_pods
+    # narrowed by any client lease. Same guard, one indirection on.
+    assert "in_hand >= self.effective_max_pods()" in blk
 
 
 def test_the_answer_names_the_model_the_client_asked_about():

tests/test_runpod_schedule_region_spend.py

@@ -136,6 +136,14 @@ def test_effective_min_pods_drops_to_zero_outside_the_window(monkeypatch):
         _client_floor = 0
         _client_floor_until = 0.0
         _client_floor_logged = -1
+        # The floor is also clamped by the CEILING in force, which a lease may
+        # narrow (/v1/runpod/scale max_pods). No lease here either: this is the
+        # schedule.
+        effective_max_pods = rw.RunpodPodPool.effective_max_pods
+        client_cap = rw.RunpodPodPool.client_cap
+        _client_cap = 0
+        _client_cap_until = 0.0
+        _client_cap_logged = -1
 
     pool = _Pool()
     monkeypatch.setattr(rw, "schedule_state", lambda cfg, now=None: {"in_window": True})

tests/test_runpod_warm_floor_respects_max_pods.py

@@ -105,7 +105,7 @@ def test_worker_uses_two_distinct_counts():
     src = open("codai/api/runpod_worker.py").read()
     assert "billing_in_hand = len(self.pods) + self._booting" in src
     assert "healthy_in_hand = len([p for p in self.pods if p.healthy]) + self._booting" in src
-    assert "max(0, int(self.mcfg.max_pods or 0) - billing_in_hand)" in src, \
+    assert "max(0, self.effective_max_pods() - billing_in_hand)" in src, \
         "the max_pods cap must be tested against pods that bill"
     assert "want = min(keep_min - healthy_in_hand," in src, \
         "the deficit must still be tested against pods that can serve"