runpod: a request may narrow which cards a lease accepts (v0.3.54)

Per model, like every other pod setting, and bounded so that it can only ever
NARROW what the operator configured.

A pool with a preference and allow_other_gpus: true takes another card when the
preferred one is not free -- usually right, occasionally exactly wrong. Observed
2026-10-13: gpu_types [NVIDIA A40] fell back to an L40S at $1.09/hr against the
A40's $0.59, correct per the configuration and useless to a caller measuring
throughput, since a number taken on a different card is not comparable with the
one it is being compared to. The only way to get the A40 was to edit the model's
configuration -- changing it for everybody, permanently, for one request.

POST /v1/runpod/scale now takes `gpu_types` (list or comma string, in preference
order) and `allow_other_gpus`. The rule is one-directional:

  - a card the operator did not list is DROPPED and named in `notes`;
  - narrowing to only cards the operator forbids is a 400 that names what IS
    allowed -- never a silent fall back to the default, which would be the
    opposite of what the caller asked for;
  - allow_other_gpus may be turned OFF by a request, never ON;
  - max_hourly_usd, min_vram_gb, cloud_types and max_pods are the operator's
    alone and are not reachable from a request at all.

Held on the LEASE, not the pool: an "A40 only" cannot outlive the 30 minutes it
asked for, releasing the lease releases its constraints, and provisioning reads
the narrowing per pod so a lapsed lease stops constraining the very next one.
There is deliberately no path from a request to a model's configuration.

Reported back in the answer and on /v1/runpod/status as `client_gpu_narrow`,
because a constraint silently dropped is worse than one refused -- the caller
would read its numbers as having come from the card it asked for -- and a pod on
an unexpected card should have a visible cause, not only a bill.

Tests: 32 new, including that the cost ceiling and the VRAM floor still refuse a
pinned card, that the operator's limits are not even mentioned in the narrowing
code, and that None (not False) is the default for both fields -- False would
silently turn every caller's fallback off.

Also fixed a source-window test that anchored on `_rank_gpus`'s full argument
list and broke the moment a parameter was added; it now anchors on the
assignment, since the arguments are not what it is testing.

Full suite: 25 failures, all pre-existing and identical to baseline.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PxYHWbCAjjjD7A1rhFKtaW

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.3.53"
+__version__ = "0.3.54"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/api/app.py

@@ -23,7 +23,7 @@ import logging
 import os
 import time
 from contextlib import asynccontextmanager
-from typing import List
+from typing import List, Optional, Union
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import FileResponse, JSONResponse
@@ -902,6 +902,21 @@ class RunpodScaleRequest(BaseModel):
                                        "waiting out scale_down_after_s. Never goes "
                                        "below the model's own min_pods, and a pod "
                                        "with requests on it drains first.")
+    gpu_types: Optional[Union[str, List[str]]] = Field(
+        None,
+        description="Cards THIS lease will accept, in preference order (a list, "
+                    "or a comma-separated string). May only NARROW the model's "
+                    "configured gpu_types: a card the operator did not list is "
+                    "ignored and reported back in `notes`, and narrowing to "
+                    "nothing allowed is a 400. Omit to use the model's own "
+                    "preference.")
+    allow_other_gpus: Optional[bool] = Field(
+        None,
+        description="false makes this lease wait for one of its gpu_types "
+                    "instead of taking whatever else fits. May only be turned "
+                    "OFF by a request: asking for true where the operator set "
+                    "false is ignored and reported. Omit to use the model's own "
+                    "setting.")
     model_config = ConfigDict(extra="allow")
 
 
@@ -922,6 +937,14 @@ async def runpod_scale(body: RunpodScaleRequest, request: Request):
     still cannot scale through the operator's own ``min_pods``, and a pod with
     requests on it drains rather than dying under them.
 
+    ``gpu_types`` and ``allow_other_gpus`` let a lease say which cards it will
+    accept. A caller measuring throughput knows something the operator cannot --
+    that a number taken on a different card is not comparable with yesterday's,
+    and that waiting for the right one beats paying a fallback's price for a
+    result it cannot use. It may only ever NARROW the model's own choice, never
+    widen it, and the narrowing dies with the lease: there is deliberately no way
+    for a request to change a model's configuration.
+
     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
@@ -930,10 +953,15 @@ async def runpod_scale(body: RunpodScaleRequest, request: Request):
     """
     from codai.api import runpod_worker
     try:
-        return runpod_worker.set_client_pods(body.model, body.pods,
-                                            teardown=bool(body.teardown))
+        return 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)
     except LookupError as exc:
         raise HTTPException(status_code=404, detail=str(exc))
+    except ValueError as exc:
+        # A narrowing that leaves nothing allowed: the caller's mistake, and it
+        # must not quietly fall back to the operator's default.
+        raise HTTPException(status_code=400, detail=str(exc))
     except PermissionError as exc:
         raise HTTPException(status_code=403, detail=str(exc))
     except Exception as exc:

codai/api/runpod_worker.py

@@ -1870,7 +1870,80 @@ def _pod_health_ok(url: str, timeout: float = 4.0, path: str = "/v1/models",
         return False
 
 
-def _rank_gpus(client, mcfg: "RunpodModelConfig", account_cfg) -> list:
+def narrow_gpu_choice(mcfg: "RunpodModelConfig", gpu_types=None,
+                      allow_other_gpus=None) -> dict:
+    """Apply a caller's GPU constraints to the operator's, and report the result.
+
+    Returns ``{"gpu_types": [...], "allow_other_gpus": bool, "notes": [...]}``.
+
+    The rule is one-directional and deliberately so: **a caller may narrow the
+    operator's choices, never widen them.** Picking the card is normally the
+    operator's business -- they set the preference, the cost ceiling and the VRAM
+    floor -- but a caller legitimately knows things the operator cannot: that
+    this batch must be comparable with yesterday's numbers, that a measurement is
+    worthless on a different card, that it would rather wait than pay a
+    fallback's price. Letting it say so costs nothing. Letting it widen would
+    hand it the operator's budget.
+
+    So:
+
+    * a requested card that the operator did not list is DROPPED, not honoured,
+      and said so in `notes`;
+    * `allow_other_gpus` may be turned OFF by a caller, never ON;
+    * `max_hourly_usd`, `min_vram_gb`, `cloud_types` and `max_pods` are the
+      operator's alone and are not reachable from here at all.
+
+    Narrowing to nothing is refused rather than silently ignored: a caller asking
+    only for cards the operator forbids has made a mistake worth hearing about,
+    and dropping to "anything" would be the opposite of what it asked.
+    """
+    configured = [str(g).strip() for g in (getattr(mcfg, "gpu_types", None) or [])
+                  if str(g).strip()]
+    allow_other = bool(getattr(mcfg, "allow_other_gpus", False))
+    notes = []
+
+    want = gpu_types
+    if isinstance(want, str):
+        want = [w.strip() for w in want.split(",") if w.strip()]
+    want = [str(w).strip() for w in (want or []) if str(w).strip()]
+
+    out_types = list(configured)
+    if want:
+        if configured:
+            # Case-insensitively, keep the caller's order: it is a preference
+            # order, and the caller's ranking is the point of asking.
+            lower = {c.lower(): c for c in configured}
+            kept, dropped = [], []
+            for w in want:
+                hit = lower.get(w.lower())
+                (kept.append(hit) if hit else dropped.append(w))
+            if dropped:
+                notes.append(
+                    f"ignored {dropped}: not in the operator's gpu_types "
+                    f"{configured}")
+            if not kept:
+                raise ValueError(
+                    f"none of the requested GPUs {want} is allowed here; this "
+                    f"model's gpu_types is {configured}")
+            out_types = kept
+        else:
+            # No operator preference means "any card within the limits", so a
+            # caller naming one is strictly more restrictive.
+            out_types = want
+
+    out_allow = allow_other
+    if allow_other_gpus is not None:
+        if bool(allow_other_gpus):
+            if not allow_other:
+                notes.append(
+                    "ignored allow_other_gpus=true: the operator set it false, "
+                    "and a request may only narrow the choice")
+        else:
+            out_allow = False
+    return {"gpu_types": out_types, "allow_other_gpus": out_allow, "notes": notes}
+
+
+def _rank_gpus(client, mcfg: "RunpodModelConfig", account_cfg, narrow=None) -> list:
     """Ranked list of GPU options across the model's allowed pools per its selection
     criteria (best first). Each item: {gpu_type_id, display_name, memory_gb,
     cloud_type, price, is_spot, bid}. Raises RunpodError if nothing fits the
@@ -1894,6 +1967,14 @@ def _rank_gpus(client, mcfg: "RunpodModelConfig", account_cfg) -> list:
                   if str(getattr(account_cfg, "default_gpu_type", "") or "").strip()
                   else []))
     allow_other = bool(getattr(mcfg, "allow_other_gpus", False))
+    # A caller's narrowed choice, if one is in force (see `narrow_gpu_choice`).
+    # Applied here rather than on the config, so it cannot outlive the lease that
+    # asked for it or leak into another caller's provisioning.
+    if narrow:
+        if narrow.get("gpu_types"):
+            prefer = list(narrow["gpu_types"])
+        if narrow.get("allow_other_gpus") is not None:
+            allow_other = bool(narrow["allow_other_gpus"])
 
     cands = []   # (pref_rank, price, is_spot, cloud_type, gpu, on_demand_price)
     for g in catalog:
@@ -2241,6 +2322,11 @@ class RunpodPodPool:
         self._client_floor = 0
         self._client_floor_until = 0.0
         self._client_floor_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`.
+        self._client_gpu_types: list = []
+        self._client_allow_other = None
         #: Pods THIS process is provisioning right now. A count, not a flag: it
         #: used to be a boolean, and because a pod takes minutes to boot, the
         #: pool could grow by at most one pod per boot however much work was
@@ -2741,7 +2827,10 @@ class RunpodPodPool:
         Retries on a different machine when a pod is accepted but never boots."""
         from codai.api.runpod_client import RunpodClient, RunpodError, pod_console_url
         client = RunpodClient(self.account)
-        ranked = _rank_gpus(client, self.mcfg, self.account)
+        # The caller's narrowing, if a lease is in force. Read per provision so
+        # a lapsed lease stops constraining the very next pod.
+        ranked = _rank_gpus(client, self.mcfg, self.account,
+                            narrow=self.client_gpu_narrow())
         # Each (card, region) pair is its own candidate, so the existing
         # capacity fallback walks regions too: no A40 in the first EU region
         # falls through to the next instead of failing the request.
@@ -3187,7 +3276,25 @@ class RunpodPodPool:
             return 0
         return min(self._client_floor, max(1, self.mcfg.max_pods))
 
-    def set_client_floor(self, pods: int) -> dict:
+    def client_gpu_narrow(self) -> dict:
+        """The GPU narrowing in force right now, or {} once the lease lapses.
+
+        Tied to the lease rather than to the pool, so a caller's "A40 only"
+        cannot outlive the 30 minutes it asked for, and the pool goes back to the
+        operator's own preference by itself. Nothing persists it: there is
+        deliberately no way for a request to change the model's configuration.
+        """
+        if self.client_floor() <= 0:
+            return {}
+        out = {}
+        if self._client_gpu_types:
+            out["gpu_types"] = list(self._client_gpu_types)
+        if self._client_allow_other is not None:
+            out["allow_other_gpus"] = bool(self._client_allow_other)
+        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.
 
         Clamped to max_pods: the configured ceiling is the budget, and a client
@@ -3202,15 +3309,43 @@ 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))
+        # 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)
         with self._cv:
             self._client_floor = granted
             self._client_floor_until = (time.time() + ttl) if granted else 0.0
             self._client_floor_logged = -1
+            if granted:
+                self._client_gpu_types = (list(narrow["gpu_types"])
+                                          if gpu_types else [])
+                self._client_allow_other = (narrow["allow_other_gpus"]
+                                            if allow_other_gpus is not None
+                                            else None)
+            else:
+                # Releasing the lease releases its constraints too, or an
+                # "A40 only" would quietly outlive the request that wanted it.
+                self._client_gpu_types = []
+                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)
-        return {"model": self.model_key, "requested": want, "granted": granted,
+        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,
                "clamped": want > cap}
+        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
+            # having come from the card it asked for.
+            out["gpu_types"] = narrow["gpu_types"]
+            out["allow_other_gpus"] = narrow["allow_other_gpus"]
+            if narrow["notes"]:
+                out["notes"] = narrow["notes"]
+        return out
 
     def effective_min_pods(self) -> int:
         """The warm floor that applies right now.
@@ -3838,11 +3973,16 @@ def pools_schedule_status() -> list:
             "client_pods": pool.client_floor(),
             "client_lease_expires_in_s": max(
                 0, int(pool._client_floor_until - time.time())) if pool.client_floor() else 0,
+            # What the lease narrowed the card choice to, if anything: a pod on
+            # an unexpected card should have a visible cause too, not only a
+            # bill.
+            "client_gpu_narrow": pool.client_gpu_narrow(),
         })
     return out
 
 
-def set_client_pods(model: str, pods: int, teardown: bool = False) -> dict:
+def set_client_pods(model: str, pods: int, teardown: bool = False,
+                    gpu_types=None, allow_other_gpus=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
@@ -3853,6 +3993,15 @@ def set_client_pods(model: str, pods: int, teardown: bool = False) -> dict:
     scale-down window: the client that just finished a batch knows the work has
     stopped, which the autoscaler can only infer slowly. It cannot go below the
     operator's warm floor, and it drains rather than dropping work.
+
+    ``gpu_types`` and ``allow_other_gpus`` constrain the cards this lease will
+    accept, and may only NARROW what the operator configured -- see
+    `narrow_gpu_choice`. They are the answer to a real need: a caller measuring
+    throughput knows that a number taken on a different card is not comparable
+    with yesterday's, and would rather wait for the right one than pay a
+    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.
     """
     key = str(model or "").strip()
     if not key:
@@ -3872,7 +4021,8 @@ def set_client_pods(model: str, pods: int, teardown: bool = False) -> dict:
     # Creating the pool is the point: asking for a warm floor on a model nobody
     # 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)
+    out = pool.set_client_floor(pods, gpu_types=gpu_types,
+                                allow_other_gpus=allow_other_gpus)
     if teardown:
         # After the floor is lowered, so the shed measures against the new one.
         out["teardown"] = pool.shed_now()

docs/orchestrator-runpod-guide.md

@@ -774,6 +774,57 @@ curl -fsS -X POST http://127.0.0.1:8776/v1/runpod/scale \
  "max_pods": 3, "expires_in_s": 1800, "clamped": false}
 ```
 
+#### Choosing the card for one lease
+
+A pool configured with a preference and `allow_other_gpus: true` will take
+another card when the preferred one is not free — which is usually what you
+want, and occasionally exactly what you do not. A caller measuring throughput
+knows something the operator cannot: that a number taken on a different card is
+not comparable with yesterday's, and that waiting for the right card beats
+paying a fallback's price for a result it cannot use.
+
+So a lease may say which cards it accepts:
+
+```bash
+curl -fsS -X POST http://127.0.0.1:8776/v1/runpod/scale \
+     -H "Authorization: Bearer sk-coderai-…" -H "Content-Type: application/json" \
+     -d '{"model": "qwen38-awq", "pods": 1,
+          "gpu_types": ["NVIDIA A40"], "allow_other_gpus": false}'
+```
+
+```json
+{"model": "qwen38-awq", "requested": 1, "granted": 1, "max_pods": 3,
+ "expires_in_s": 1800, "clamped": false,
+ "gpu_types": ["NVIDIA A40"], "allow_other_gpus": false}
+```
+
+The rule is one-directional: **a request may narrow the operator's choice, never
+widen it.**
+
+| asked | result |
+|---|---|
+| a card in the model's `gpu_types` | honoured, in the order the caller listed |
+| a card the model does **not** list | dropped, and named in `notes` |
+| only cards the model forbids | **400**, naming what is allowed — never a silent fall back to the default |
+| `allow_other_gpus: false` | honoured: this lease waits rather than falling back |
+| `allow_other_gpus: true` where the model says false | ignored, and said so in `notes` |
+| `max_hourly_usd`, `min_vram_gb`, `cloud_types`, `max_pods` | not reachable from a request at all |
+
+Two properties worth knowing:
+
+- **It dies with the lease.** The narrowing is held on the lease, not the pool,
+  so an "A40 only" cannot outlive the thirty minutes it asked for, and the pool
+  returns to your preference by itself. There is deliberately **no way for a
+  request to change a model's configuration** — that stays yours.
+- **A dropped constraint is reported, never silent.** Otherwise a caller reads
+  its numbers as having come from the card it asked for. `/v1/runpod/status`
+  shows any narrowing in force as `client_gpu_narrow`, so a pod on an unexpected
+  card has a visible cause and not only a bill.
+
+It is per model, like every other pod setting: the gate is the same
+`allow_client_scale`, and the bounds are that model's own `gpu_types`,
+`allow_other_gpus` and cost ceiling.
+
 It is a **lease**, bounded three ways, because a client that can raise the floor
 can raise the bill:
 

tests/test_runpod_placement.py

@@ -160,7 +160,10 @@ def test_candidates_are_expanded_across_regions_for_the_capacity_fallback():
     """The existing retry walks `ranked` on a capacity miss; pairing each card
     with each region makes it walk regions too, with no new control flow."""
     src = (ROOT / "codai/api/runpod_worker.py").read_text()
-    blk = src.split("ranked = _rank_gpus(client, self.mcfg, self.account)")[1][:700]
+    # Anchored on the assignment alone: the call's ARGUMENTS are not what this
+    # test is about, and spelling them out made it fail the day a parameter was
+    # added and wrapped the line.
+    blk = src.split("ranked = _rank_gpus(")[1][:900]
     assert 'dict(sel, _dc=dc) for sel in ranked for dc in dcs' in blk
     assert "if len(dcs) > 1:" in blk
 

tests/test_runpod_request_gpu_choice.py

+"""A request may narrow which cards a lease accepts — never widen.
+
+Picking the card is normally the operator's business: they set the preference,
+the cost ceiling, the VRAM floor and the pools. But a caller knows things the
+operator cannot — that this batch has to be comparable with yesterday's numbers,
+that a measurement on a different card is worthless, that it would rather wait
+than pay a fallback's price. Letting it say so costs nothing; letting it WIDEN
+would hand it the operator's budget.
+
+The case that prompted this: a pool configured `gpu_types: [NVIDIA A40]` with
+`allow_other_gpus: true` fell back to an L40S at $1.09/hr against the A40's
+$0.59 when no A40 was free. Correct per the configuration, and useless to a
+caller measuring throughput. Before this the only way to get the A40 was to edit
+the model's configuration — which changes it for everybody, permanently, for the
+sake of one request.
+
+Per MODEL, not per engine: every pool goes through the same resolver, and
+`gpu_types`/`allow_other_gpus` are per-model settings already.
+"""
+import time
+import types
+
+import pytest
+
+from codai.api import runpod_worker as rw
+
+
+def _mcfg(gpu_types=None, allow_other=False, **kw):
+    cfg = rw.RunpodModelConfig()
+    cfg.gpu_types = list(gpu_types or [])
+    cfg.allow_other_gpus = bool(allow_other)
+    for k, v in kw.items():
+        setattr(cfg, k, v)
+    return cfg
+
+
+# --- narrowing ---------------------------------------------------------------
+
+
+def test_nothing_requested_leaves_the_operators_choice_alone():
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40"], allow_other=True))
+    assert out["gpu_types"] == ["NVIDIA A40"]
+    assert out["allow_other_gpus"] is True
+    assert out["notes"] == []
+
+
+def test_a_request_can_pin_one_of_the_allowed_cards():
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40", "NVIDIA L40S"], allow_other=True),
+                               gpu_types=["NVIDIA A40"])
+    assert out["gpu_types"] == ["NVIDIA A40"]
+
+
+def test_a_request_can_refuse_the_fallback():
+    """The actual fix for the $1.09 L40S: wait for the A40 rather than take
+    whatever else fits."""
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40"], allow_other=True),
+                               gpu_types=["NVIDIA A40"], allow_other_gpus=False)
+    assert out["gpu_types"] == ["NVIDIA A40"]
+    assert out["allow_other_gpus"] is False
+
+
+def test_the_requests_order_is_kept_because_it_is_a_preference():
+    out = rw.narrow_gpu_choice(
+        _mcfg(["NVIDIA A40", "NVIDIA L40S", "NVIDIA A100"], allow_other=True),
+        gpu_types=["NVIDIA A100", "NVIDIA A40"])
+    assert out["gpu_types"] == ["NVIDIA A100", "NVIDIA A40"]
+
+
+def test_a_comma_string_is_accepted():
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40", "NVIDIA L40S"]),
+                               gpu_types="NVIDIA L40S, NVIDIA A40")
+    assert out["gpu_types"] == ["NVIDIA L40S", "NVIDIA A40"]
+
+
+def test_case_does_not_matter_but_the_operators_spelling_is_published():
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40"]), gpu_types=["nvidia a40"])
+    assert out["gpu_types"] == ["NVIDIA A40"]
+
+
+# --- what a request must NOT be able to do -----------------------------------
+
+
+def test_a_card_the_operator_did_not_list_is_dropped_and_reported():
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40"], allow_other=True),
+                               gpu_types=["NVIDIA A40", "NVIDIA H100 NVL"])
+    assert out["gpu_types"] == ["NVIDIA A40"]
+    assert any("H100" in n for n in out["notes"]), out["notes"]
+
+
+def test_narrowing_to_nothing_allowed_is_refused_not_ignored():
+    """Falling back to the operator's default would be the OPPOSITE of what the
+    caller asked for, and it would do so silently."""
+    with pytest.raises(ValueError) as e:
+        rw.narrow_gpu_choice(_mcfg(["NVIDIA A40"]), gpu_types=["NVIDIA H100 NVL"])
+    assert "A40" in str(e.value), "the error does not say what IS allowed"
+
+
+def test_a_request_cannot_turn_the_fallback_on():
+    out = rw.narrow_gpu_choice(_mcfg(["NVIDIA A40"], allow_other=False),
+                               allow_other_gpus=True)
+    assert out["allow_other_gpus"] is False
+    assert any("narrow" in n for n in out["notes"]), out["notes"]
+
+
+def test_with_no_operator_preference_a_request_may_still_pin_a_card():
+    """No configured preference means "any card within the limits", so naming
+    one is strictly more restrictive — and the ceiling still binds."""
+    out = rw.narrow_gpu_choice(_mcfg([], allow_other=True),
+                               gpu_types=["NVIDIA A40"])
+    assert out["gpu_types"] == ["NVIDIA A40"]
+
+
+@pytest.mark.parametrize("field", ["max_hourly_usd", "min_vram_gb", "max_pods",
+                                   "cloud_types", "allow_spot"])
+def test_the_operators_limits_are_not_reachable_from_a_request(field):
+    """Only the card choice is negotiable. The cost ceiling, the VRAM floor, the
+    pools and the pod cap are the operator's alone."""
+    import inspect
+    src = inspect.getsource(rw.narrow_gpu_choice)
+    body = src.split('"""')[2]  # past the docstring
+    assert field not in body, f"narrow_gpu_choice touches {field}"
+
+
+# --- the lease carries it, and it dies with the lease ------------------------
+
+
+def _pool(mcfg):
+    pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
+    import threading
+    pool._cv = threading.Condition()
+    pool.mcfg = mcfg
+    pool.model_key = "m"
+    pool.pods = []
+    pool._client_floor = 0
+    pool._client_floor_until = 0.0
+    pool._client_floor_logged = -1
+    pool._client_gpu_types = []
+    pool._client_allow_other = None
+    pool._release_pending = False
+    return pool
+
+
+def test_the_lease_reports_what_it_granted():
+    pool = _pool(_mcfg(["NVIDIA A40", "NVIDIA L40S"], allow_other=True, max_pods=4))
+    out = pool.set_client_floor(1, gpu_types=["NVIDIA A40"],
+                                allow_other_gpus=False)
+    assert out["granted"] == 1
+    assert out["gpu_types"] == ["NVIDIA A40"]
+    assert out["allow_other_gpus"] is False
+
+
+def test_a_dropped_card_is_reported_in_the_answer():
+    """A constraint silently dropped is worse than one refused: the caller would
+    read the numbers as having come from the card it asked for."""
+    pool = _pool(_mcfg(["NVIDIA A40"], allow_other=True, max_pods=4))
+    out = pool.set_client_floor(1, gpu_types=["NVIDIA A40", "NVIDIA H100 NVL"])
+    assert out["gpu_types"] == ["NVIDIA A40"]
+    assert "notes" in out and any("H100" in n for n in out["notes"])
+
+
+def test_the_narrowing_applies_while_the_lease_lives():
+    pool = _pool(_mcfg(["NVIDIA A40", "NVIDIA L40S"], allow_other=True, max_pods=4))
+    pool.set_client_floor(1, gpu_types=["NVIDIA A40"], allow_other_gpus=False)
+    assert pool.client_gpu_narrow() == {"gpu_types": ["NVIDIA A40"],
+                                        "allow_other_gpus": False}
+
+
+def test_the_narrowing_dies_WITH_the_lease():
+    """Otherwise one request's "A40 only" quietly becomes the pool's policy."""
+    pool = _pool(_mcfg(["NVIDIA A40", "NVIDIA L40S"], allow_other=True, max_pods=4))
+    pool.set_client_floor(1, gpu_types=["NVIDIA A40"], allow_other_gpus=False)
+    pool._client_floor_until = time.time() - 1          # lease lapsed
+    assert pool.client_gpu_narrow() == {}
+
+
+def test_releasing_the_lease_releases_its_constraints():
+    pool = _pool(_mcfg(["NVIDIA A40", "NVIDIA L40S"], allow_other=True, max_pods=4))
+    pool.set_client_floor(1, gpu_types=["NVIDIA A40"], allow_other_gpus=False)
+    pool.set_client_floor(0)
+    assert pool.client_gpu_narrow() == {}
+    assert pool._client_gpu_types == [] and pool._client_allow_other is None
+
+
+def test_a_plain_lease_constrains_nothing():
+    pool = _pool(_mcfg(["NVIDIA A40"], allow_other=True, max_pods=4))
+    out = pool.set_client_floor(2)
+    assert pool.client_gpu_narrow() == {}
+    assert "gpu_types" not in out, "it reported a constraint nobody asked for"
+
+
+def test_an_impossible_narrowing_raises_from_the_lease_too():
+    pool = _pool(_mcfg(["NVIDIA A40"], allow_other=True, max_pods=4))
+    with pytest.raises(ValueError):
+        pool.set_client_floor(1, gpu_types=["NVIDIA H100 NVL"])
+
+
+# --- ranking honours it ------------------------------------------------------
+
+
+CATALOG = [
+    {"id": "NVIDIA A40", "display_name": "A40", "memory_gb": 48,
+     "secure_price": 0.59, "community_price": 0.4, "spot_price": None},
+    {"id": "NVIDIA L40S", "display_name": "L40S", "memory_gb": 48,
+     "secure_price": 1.09, "community_price": 0.9, "spot_price": None},
+]
+
+
+def _client():
+    return types.SimpleNamespace(list_gpu_types=lambda: list(CATALOG))
+
+
+def _acct():
+    return types.SimpleNamespace(default_gpu_type="")
+
+
+def test_ranking_without_narrowing_allows_the_fallback():
+    cfg = _mcfg(["NVIDIA A40"], allow_other=True, cloud_types=["SECURE"],
+                max_hourly_usd=1.2, min_vram_gb=40, gpu_count=1,
+                allow_spot=False, gpu_type="")
+    ids = [g["gpu_type_id"] for g in rw._rank_gpus(_client(), cfg, _acct())]
+    assert "NVIDIA A40" in ids and "NVIDIA L40S" in ids
+    assert ids[0] == "NVIDIA A40", "the operator's preference lost its rank"
+
+
+def test_ranking_with_narrowing_excludes_the_fallback():
+    """The point of the whole feature: no L40S on offer, so a capacity miss
+    waits for an A40 instead of silently costing twice as much."""
+    cfg = _mcfg(["NVIDIA A40"], allow_other=True, cloud_types=["SECURE"],
+                max_hourly_usd=1.2, min_vram_gb=40, gpu_count=1,
+                allow_spot=False, gpu_type="")
+    ranked = rw._rank_gpus(_client(), cfg, _acct(),
+                           narrow={"gpu_types": ["NVIDIA A40"],
+                                   "allow_other_gpus": False})
+    assert [g["gpu_type_id"] for g in ranked] == ["NVIDIA A40"]
+
+
+def test_narrowing_cannot_defeat_the_cost_ceiling():
+    """A request pinning an expensive card it is allowed to name still cannot
+    spend past the operator's ceiling."""
+    cfg = _mcfg(["NVIDIA A40", "NVIDIA L40S"], allow_other=False,
+                cloud_types=["SECURE"], max_hourly_usd=0.80, min_vram_gb=40,
+                gpu_count=1, allow_spot=False, gpu_type="")
+    from codai.api.runpod_client import RunpodError
+    with pytest.raises(RunpodError):
+        rw._rank_gpus(_client(), cfg, _acct(),
+                      narrow={"gpu_types": ["NVIDIA L40S"],
+                              "allow_other_gpus": False})
+
+
+def test_narrowing_cannot_defeat_the_vram_floor():
+    cfg = _mcfg(["NVIDIA A40"], allow_other=False, cloud_types=["SECURE"],
+                max_hourly_usd=5.0, min_vram_gb=80, gpu_count=1,
+                allow_spot=False, gpu_type="")
+    from codai.api.runpod_client import RunpodError
+    with pytest.raises(RunpodError):
+        rw._rank_gpus(_client(), cfg, _acct(),
+                      narrow={"gpu_types": ["NVIDIA A40"],
+                              "allow_other_gpus": False})
+
+
+def test_provisioning_reads_the_narrowing_per_pod():
+    """Read per provision, not cached, so a lapsed lease stops constraining the
+    very next pod."""
+    import inspect
+    src = inspect.getsource(rw.RunpodPodPool._provision_one)
+    assert "narrow=self.client_gpu_narrow()" in src
+
+
+# --- the HTTP surface --------------------------------------------------------
+
+
+def test_the_request_model_accepts_both_fields():
+    from codai.api.app import RunpodScaleRequest
+    b = RunpodScaleRequest(model="m", pods=1, gpu_types=["NVIDIA A40"],
+                           allow_other_gpus=False)
+    assert b.gpu_types == ["NVIDIA A40"] and b.allow_other_gpus is False
+
+
+def test_both_fields_are_optional_and_default_to_the_model():
+    """None must mean "use the model's own setting" — False would silently turn
+    every caller's fallback off."""
+    from codai.api.app import RunpodScaleRequest
+    b = RunpodScaleRequest(model="m", pods=1)
+    assert b.gpu_types is None and b.allow_other_gpus is None
+
+
+def test_a_comma_string_is_accepted_over_http_too():
+    from codai.api.app import RunpodScaleRequest
+    b = RunpodScaleRequest(model="m", pods=1, gpu_types="NVIDIA A40,NVIDIA L40S")
+    assert isinstance(b.gpu_types, str)
+    assert rw.narrow_gpu_choice(
+        _mcfg(["NVIDIA A40", "NVIDIA L40S"]), gpu_types=b.gpu_types
+    )["gpu_types"] == ["NVIDIA A40", "NVIDIA L40S"]
+
+
+def test_an_impossible_narrowing_is_a_400_not_a_500():
+    import inspect
+    from codai.api import app as app_mod
+    src = inspect.getsource(app_mod.runpod_scale)
+    assert "except ValueError" in src
+    i, j = src.index("except ValueError"), src.index("except Exception")
+    assert "400" in src[i:j], "a caller's mistake is reported as a server error"
+
+
+def test_the_status_page_shows_a_leases_narrowing():
+    """A pod on an unexpected card should have a visible cause, not only a bill."""
+    import inspect
+    src = inspect.getsource(rw.pools_schedule_status)
+    assert '"client_gpu_narrow": pool.client_gpu_narrow()' in src