longcat: VRAM accounting, a draining releaser, and video in the front queue (v0.2.32)

Step 3: make LongCat a citizen of the memory and swap machinery rather than a model that
merely runs. Three gaps had to close.

1. THE ESTIMATE. _get_model_used_vram_gb had no branch for a `longcat:` key, so it fell
   through to scanning the checkpoint — ~83 GB across variants, which would have asked
   the manager to evict the whole box on every request. Added beside the ds4/colibri/vllm
   early returns: measured if known, else the starting estimate.

2. THE MEASUREMENT. No official VRAM figure exists for LongCat and the real peak moves
   with the variant, the stage and the offload mode, so the constant is a starting point
   and the service reports what it actually reached. It has to come from inside the
   worker: the pipeline loads lazily on the first request, so a host-side free-VRAM delta
   would also be counting whatever else was loading on the card. torch's
   max_memory_reserved is this process's own arenas — what the card genuinely cannot give
   to anything else. The HIGH-WATER mark is kept, because a draft run touches far less
   than a 720p refinement and reserving the smaller figure would under-evict later, and
   it is keyed per config_id/variant so an fp8 sibling cannot inherit a bf16 measurement.

3. GIVING THE CARD BACK. register_external_vram_releaser with the documented
   fn(needed_gb) -> float contract (not voice_clone's zero-argument one, which raises
   TypeError inside eviction and is therefore never reclaimed). The drain matters more
   here than anywhere: acquire_stt_backend POPS the model pool, so _is_key_busy is
   permanently False for this model and Pass 1 of release_idle_vram would tear the
   subprocess down mid-generation and lose a multi-minute request. The releaser waits for
   in-flight work, bounded by longcat.evict_drain_timeout_s (300 s).

Also: video can now take a slot in the front per-model queue. Every acquire site was
guarded by `_task_kind(path) == "text"`, so max_instances on a video model did nothing at
the front and nothing bounded concurrent generations but the engine and the swap gate.
_queues_at_front() enables it for video when — and only when — the operator has set
max_instances: doing it unconditionally would change behaviour for every existing video
model and start returning 503 at queue_max_size, which is not something to impose on
deployments that never asked. Applied at all four sites (direct, keepalive, both
brokered), and the `is_text` local it replaced was left dead, so it is gone.

21 tests, including the one that matters most: an eviction arriving mid-generation waits
and the request still completes.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

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

codai/admin/routes.py

@@ -3789,6 +3789,7 @@ def build_settings_dict(c, gpu_cards):
             "gpu": c.longcat.gpu,
             "attention": c.longcat.attention,
             "ready_timeout": c.longcat.ready_timeout,
+            "evict_drain_timeout_s": c.longcat.evict_drain_timeout_s,
             "extra_args": c.longcat.extra_args,
             "extra_env": c.longcat.extra_env,
         },
@@ -4432,6 +4433,11 @@ async def api_save_settings(request: Request, username: str = Depends(require_ad
                 lc.ready_timeout = max(0.0, float(d.get("ready_timeout") or 0.0))
             except (TypeError, ValueError):
                 pass
+        if "evict_drain_timeout_s" in d:
+            try:
+                lc.evict_drain_timeout_s = max(0.0, float(d.get("evict_drain_timeout_s") or 0.0))
+            except (TypeError, ValueError):
+                pass
         if lc.attention not in ("xformers", "flash", "sdpa"):
             lc.attention = "xformers"
     if "vllm" in data:

codai/api/longcat_worker.py

@@ -75,6 +75,13 @@ DEFAULT_VRAM_GB = 24.0
 
 _lock = threading.RLock()
 _services: dict = {}          # key -> {"proc","port","url","meta"}
+# In-flight generations per service, and an event that is SET while nothing is running.
+# The model manager's own busy signal cannot see this work (acquire_stt_backend pops the
+# model pool, so _is_key_busy is always False for us), so the releaser does its own drain.
+_inflight: dict = {}
+_idle = threading.Event()
+_idle.set()
+_releaser_registered = False
 
 _common = None
 
@@ -476,6 +483,7 @@ def acquire(model_name: str, config: dict = None) -> LongcatHandle:
     model_path = resolve_model_path(model_name, cfg)
     key = service_key(model_path, cfg)
 
+    register_releaser()                          # both directions, from the first use
     url = ensure_service(model_path, cfg)        # idempotent; re-spawns on a new port
 
     def _loader():
@@ -490,6 +498,30 @@ def acquire(model_name: str, config: dict = None) -> LongcatHandle:
     return handle
 
 
+def _record_measurement(key: str, vram_gb: float) -> None:
+    """Write a measured footprint back, so the next reservation is not an estimate.
+
+    The 13.6B DiT is ~27 GB at bf16 and the reported peak for a full profile is ~41.6 GB,
+    but no official figure exists and it moves with the variant, the stage and the offload
+    mode. So the number is measured per service key — which includes config_id/variant,
+    because sibling configs are different residencies and an fp8 entry must not inherit a
+    bf16 measurement."""
+    if not vram_gb or vram_gb <= 0:
+        return
+    try:
+        from codai.models.manager import multi_model_manager
+        prev = float(multi_model_manager._measured_vram_gb.get(key) or 0.0)
+        # Keep the HIGH-WATER mark: a draft-preset run touches far less of the card than
+        # a 720p refinement, and reserving the smaller figure would under-evict for the
+        # bigger one later.
+        if vram_gb > prev + 0.05:
+            multi_model_manager._measured_vram_gb[key] = float(vram_gb)
+            print(f"[longcat] measured {vram_gb:.2f} GB for {key}"
+                  + (f" (was {prev:.2f})" if prev else ""), flush=True)
+    except Exception as e:
+        print(f"[longcat] could not record the measurement: {e}", flush=True)
+
+
 def generate(model_name: str, payload: dict, config: dict = None,
              timeout: float = 14400.0) -> dict:
     """Run one generation on the service. Returns its JSON response.
@@ -497,8 +529,18 @@ def generate(model_name: str, payload: dict, config: dict = None,
     The default timeout is four hours: a 720p refinement over many segments is measured in
     minutes per segment, and a timeout that fires mid-generation wastes all of it."""
     import requests
-    handle = acquire(model_name, config)
-    resp = requests.post(handle.url + "/generate", json=payload, timeout=timeout)
+    cfg = config or {}
+    handle = acquire(model_name, cfg)
+    key = handle.key
+    with _lock:
+        _inflight[key] = _inflight.get(key, 0) + 1
+    try:
+        resp = requests.post(handle.url + "/generate", json=payload, timeout=timeout)
+    finally:
+        with _lock:
+            _inflight[key] = max(0, _inflight.get(key, 1) - 1)
+            if _inflight[key] == 0:
+                _idle.set()
     if not resp.ok:
         try:
             detail = resp.json().get("error") or resp.text[:800]
@@ -508,8 +550,76 @@ def generate(model_name: str, payload: dict, config: dict = None,
     data = resp.json()
     if data.get("error"):
         raise RuntimeError(f"LongCat generation failed: {data['error']}")
+    _record_measurement(key, float(data.get("vram_gb") or 0.0))
     return data
 
 
+# ── eviction: giving the card back ───────────────────────────────────────────
+
+def _drain_timeout_s() -> float:
+    sec = _cfg_section()
+    try:
+        return max(0.0, float(getattr(sec, "evict_drain_timeout_s", 0) or 0)) or 300.0
+    except Exception:
+        return 300.0
+
+
+def release_vram(needed_gb: float = 999.0) -> float:
+    """External VRAM releaser: stop the services, letting in-flight work finish first.
+
+    Registered with the model manager so another model can take the card — the reverse
+    direction of acquire()'s eviction. The contract is ``fn(needed_gb) -> float``: safe to
+    call with nothing to free, and it must not raise (the manager swallows exceptions into
+    a warning, which is how voice_clone's zero-argument releaser has been silently failing).
+
+    The wait matters more here than anywhere else. A segment is ~80 new frames and minutes
+    of work, and the manager's own busy signal cannot see any of it: acquire_stt_backend
+    pops the model pool, so ``_is_key_busy`` is permanently False for this model and Pass 1
+    of release_idle_vram would tear the subprocess down mid-generation, losing the whole
+    request. Upstream already calls torch_gc() between segments, so a boundary exists;
+    until the segment loop lands, draining means waiting for the in-flight request."""
+    with _lock:
+        keys = list(_services.keys())
+        busy = sum(_inflight.get(k, 0) for k in keys)
+    if not keys:
+        return 0.0
+    if busy:
+        budget = _drain_timeout_s()
+        print(f"[longcat] {busy} generation(s) in flight — waiting up to {budget:.0f}s "
+              f"for a clean handover before releasing the card", flush=True)
+        _idle.clear()
+        if not _idle.wait(budget):
+            print(f"[longcat] still busy after {budget:.0f}s — releasing anyway; the "
+                  f"in-flight request will fail", flush=True)
+
+    freed = 0.0
+    try:
+        from codai.models.manager import multi_model_manager
+        for key in keys:
+            freed += float(multi_model_manager._measured_vram_gb.get(key) or 0.0) \
+                or DEFAULT_VRAM_GB
+    except Exception:
+        freed = float(len(keys)) * DEFAULT_VRAM_GB
+    for key in keys:
+        stop_service(key)
+    if freed:
+        print(f"[longcat] released ~{freed:.1f} GB (services stopped for VRAM "
+              f"eviction); they restart on the next request", flush=True)
+    return freed
+
+
+def register_releaser() -> None:
+    """Make LongCat's VRAM reclaimable by the model manager. Idempotent."""
+    global _releaser_registered
+    if _releaser_registered:
+        return
+    try:
+        from codai.models.manager import multi_model_manager
+        multi_model_manager.register_external_vram_releaser(release_vram)
+        _releaser_registered = True
+    except Exception as e:
+        print(f"[longcat] could not register the VRAM releaser: {e}", flush=True)
+
+
 import atexit as _atexit
 _atexit.register(stop_all)

codai/config.py

@@ -785,6 +785,11 @@ class LongcatConfig:
     # installs. "flash" assumes you installed flash-attn into the venv yourself.
     attention: str = "xformers"         # xformers | flash | sdpa
     ready_timeout: float = 1800.0       # a 13.6B DiT load is not quick
+    # When another model needs the card, how long to let an in-flight generation finish
+    # before stopping the service anyway. A segment is ~80 new frames and minutes of
+    # work, and the manager's own busy signal cannot see it (the worker holds no model
+    # pool), so this is what makes the handover a clean swap instead of a lost request.
+    evict_drain_timeout_s: float = 300.0
     extra_args: str = ""                # appended to the service command line
     extra_env: str = ""                 # KEY=VALUE pairs for the service process
 
@@ -1429,6 +1434,7 @@ class ConfigManager:
                 "gpu": self.config.longcat.gpu,
                 "attention": self.config.longcat.attention,
                 "ready_timeout": self.config.longcat.ready_timeout,
+                "evict_drain_timeout_s": self.config.longcat.evict_drain_timeout_s,
                 "extra_args": self.config.longcat.extra_args,
                 "extra_env": self.config.longcat.extra_env,
             },

codai/frontproxy/app.py

@@ -393,7 +393,7 @@ class FrontProxy:
         # direct proxy path, so brokered and direct requests share one queue.
         _qkey = None
         if (method.upper() == "POST" and _router.is_inference_path(path)
-                and self._task_kind(path) == "text"
+                and self._queues_at_front(path, model)
                 and not self._computed_off_this_gpu(engine, model, path)):
             _qkey = self._queue_key(model)
             try:
@@ -547,7 +547,7 @@ class FrontProxy:
             pass
         _qkey = None
         if (method.upper() == "POST" and _is_infer
-                and self._task_kind(path) == "text"
+                and self._queues_at_front(path, model)
                 and not self._computed_off_this_gpu(engine, model, path)):
             _qkey = self._queue_key(model)
             try:
@@ -1097,6 +1097,28 @@ class FrontProxy:
         except Exception:
             return None
 
+    def _queues_at_front(self, path: str, model) -> bool:
+        """Whether this request takes a slot in the front per-model queue.
+
+        Text always has, and video never did: every acquire site was guarded by
+        `_task_kind(path) == "text"`, so `max_instances` configured on a video model had
+        no effect at the front and nothing bounded concurrent generations but the engine
+        itself and the swap gate.
+
+        Video now queues when — and only when — the operator has set `max_instances` on
+        the entry. Queueing it unconditionally would change behaviour for every existing
+        video model and start returning 503 at `queue_max_size`, which is not something
+        to impose on deployments that never asked for it. Setting the field IS the ask."""
+        kind = self._task_kind(path)
+        if kind == "text":
+            return True
+        if kind != "video":
+            return False
+        try:
+            return bool(self._model_info(model).get("max_instances"))
+        except Exception:
+            return False
+
     def _swap_owner_key(self, engine, model: Optional[str], path: Optional[str] = None) -> str:
         """The model identity that determines GPU residency for the swap gate. Same
         model → same owner (runs free); different model → a swap. Falls back to the
@@ -1998,7 +2020,6 @@ class FrontProxy:
         _MAX_TRIES = 6
         _RETRY = 5.0
         _RETRY_STATUS = {425, 429, 500, 502, 503, 504}
-        is_text = self._task_kind(path) == "text"
 
         async def _gen():
             _qkey = None
@@ -2026,7 +2047,8 @@ class FrontProxy:
                         except _asyncio.TimeoutError:
                             yield _ka("waiting for GPU (another model is finishing)")
                 # 1. Front per-model queue slot (text only) — keepalive while waiting.
-                if is_text and not self._computed_off_this_gpu(engine, model, path):
+                if (self._queues_at_front(path, model)
+                        and not self._computed_off_this_gpu(engine, model, path)):
                     _qkey = self._queue_key(model)
                     _slot_held = False
                     # `on_busy`: if there is no free local slot RIGHT NOW, burst
@@ -2459,7 +2481,7 @@ class FrontProxy:
         # audio, embeddings…) pass through unqueued.
         _qkey = None
         if (method == "POST" and _router.is_inference_path(path)
-                and self._task_kind(path) == "text"
+                and self._queues_at_front(path, model)
                 and not self._computed_off_this_gpu(engine, model, path)):
             _qkey = self._queue_key(model)
             _slot_held = False

codai/models/manager.py

@@ -3985,6 +3985,24 @@ class MultiModelManager:
         # ~128 GB on every request (needless churn) and mis-message CPU/disk offload.
         # Use the measured value if we have one, else a modest fixed reserve.
         try:
+            # LongCat-Video runs in an isolated-venv subprocess, so there is no torch
+            # object here to size and the checkpoint on disk is ~83 GB across variants —
+            # nothing like the footprint. Its key is longcat:<path>#<config_id|variant>,
+            # and the measurement the worker writes back after a load is the only honest
+            # figure: the 13.6B DiT is ~27 GB at bf16, but the real peak depends on the
+            # variant (bf16/fp8/int8), the stage and the offload mode, and no official
+            # number exists. Until measured, reserve the starting estimate.
+            if str(model_key or "").startswith("longcat:"):
+                measured = self._measured_vram_gb.get(model_key)
+                if not measured and resolved_name:
+                    measured = self._measured_vram_gb.get(resolved_name)
+                if measured:
+                    return float(measured)
+                try:
+                    from codai.api.longcat_worker import DEFAULT_VRAM_GB
+                    return float(DEFAULT_VRAM_GB)
+                except Exception:
+                    return 24.0
             if ds4_should_handle(model_key) or (resolved_name and ds4_should_handle(resolved_name)):
                 measured = self._measured_vram_gb.get(model_key)
                 if not measured and resolved_name:

tests/test_longcat_orchestration.py

+"""Step 3: LongCat as a citizen of the VRAM and swap machinery.
+
+The operator's requirement is that a newly supported model be "correctly orchestrated like
+any other, with model swap, memory management, scalability if configured". Three gaps had
+to be closed for that to be true of an isolated-venv video model:
+
+1. the VRAM estimator had no idea what a `longcat:` key costs — it would have fallen
+   through to scanning a ~83 GB checkpoint directory;
+2. nothing measured the real footprint, and no official figure exists;
+3. the manager's busy signal is permanently False for this model (acquire_stt_backend pops
+   the model pool), so eviction would tear a multi-minute generation down mid-flight;
+   and video bypassed the front queue entirely, so `max_instances` did nothing.
+"""
+
+import sys
+import threading
+import types
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.api import longcat_worker as W
+from codai.config import LongcatConfig
+from codai.frontproxy.app import FrontProxy
+from codai.models.manager import multi_model_manager as MM
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    """Each test starts with no services and no measurements of its own."""
+    W._services.clear()
+    W._inflight.clear()
+    W._idle.set()
+    for k in [k for k in MM._measured_vram_gb if str(k).startswith("longcat:")]:
+        MM._measured_vram_gb.pop(k, None)
+    yield
+    W._services.clear()
+    W._inflight.clear()
+    W._idle.set()
+
+
+# ---------------------------------------------------------------- the estimate
+def test_an_unmeasured_model_reserves_the_starting_estimate():
+    assert MM._get_model_used_vram_gb("longcat:/w/lc#bf16") == W.DEFAULT_VRAM_GB
+
+
+def test_a_measurement_wins_over_the_estimate():
+    MM._measured_vram_gb["longcat:/w/lc#fp8"] = 13.7
+    assert MM._get_model_used_vram_gb("longcat:/w/lc#fp8") == 13.7
+
+
+def test_the_checkpoint_on_disk_is_never_used_as_the_footprint(tmp_path):
+    """The repo is ~83 GB across variants; reporting that would ask the manager to evict
+    the whole box on every request."""
+    big = tmp_path / "dit"
+    big.mkdir()
+    (big / "model.safetensors").write_bytes(b"0" * (4 * 1024 * 1024))
+    est = MM._get_model_used_vram_gb(f"longcat:{tmp_path}#bf16")
+    assert est == W.DEFAULT_VRAM_GB          # not derived from the files
+
+
+def test_the_estimate_is_only_a_starting_point():
+    """No official VRAM figure exists for LongCat and it moves with variant, stage and
+    offload, so the constant must not pretend to be a measurement."""
+    assert W.DEFAULT_VRAM_GB > 0
+    src = (ROOT / "codai/api/longcat_worker.py").read_text()
+    assert "STARTING POINT" in src or "starting estimate" in src
+
+
+# ---------------------------------------------------------------- measurement
+def test_a_reported_peak_is_recorded():
+    W._record_measurement("longcat:/w/lc#bf16", 31.4)
+    assert MM._measured_vram_gb["longcat:/w/lc#bf16"] == 31.4
+
+
+def test_the_high_water_mark_is_kept():
+    """A draft run touches far less of the card than a 720p refinement; reserving the
+    smaller figure would under-evict for the bigger one later."""
+    W._record_measurement("longcat:/w/lc#bf16", 14.0)
+    W._record_measurement("longcat:/w/lc#bf16", 31.4)
+    W._record_measurement("longcat:/w/lc#bf16", 12.0)
+    assert MM._measured_vram_gb["longcat:/w/lc#bf16"] == 31.4
+
+
+def test_a_zero_measurement_is_ignored():
+    """A CPU-only or failed run reports 0.0; recording it would disable eviction."""
+    W._record_measurement("longcat:/w/lc#bf16", 0.0)
+    assert "longcat:/w/lc#bf16" not in MM._measured_vram_gb
+
+
+def test_measurements_are_per_config_not_per_path():
+    """Sibling configs are different residencies — an fp8 entry must not inherit a bf16
+    measurement."""
+    W._record_measurement("longcat:/w/lc#bf16", 31.4)
+    assert MM._get_model_used_vram_gb("longcat:/w/lc#fp8") == W.DEFAULT_VRAM_GB
+
+
+def test_the_service_reports_reserved_not_allocated():
+    """The caching allocator's arenas are what the card cannot give to anything else."""
+    svc = (ROOT / "tools/longcat_service.py").read_text()
+    assert "max_memory_reserved" in svc
+
+
+# ---------------------------------------------------------------- the releaser
+def test_the_releaser_matches_the_manager_contract():
+    """fn(needed_gb) -> float, safe with nothing to free. voice_clone's takes no argument
+    and raises TypeError inside eviction, which is why it is never reclaimed."""
+    import inspect
+    sig = inspect.signature(W.release_vram)
+    assert len(sig.parameters) == 1
+    assert W.release_vram(999.0) == 0.0          # nothing running
+
+
+def test_it_is_registered_with_the_manager(monkeypatch):
+    registered = []
+    monkeypatch.setattr(MM, "register_external_vram_releaser",
+                        lambda fn: registered.append(fn))
+    W._releaser_registered = False
+    W.register_releaser()
+    assert W.release_vram in registered
+    W.register_releaser()                        # idempotent
+    assert len(registered) == 1
+
+
+def test_registration_is_both_directions(monkeypatch):
+    """acquire() evicts others for us; the releaser lets others evict us. A model that
+    only does the first is not orchestrated like any other."""
+    src = (ROOT / "codai/api/longcat_worker.py").read_text()
+    assert "register_releaser()" in src.split("def acquire(")[1][:800]
+
+
+def test_an_idle_service_is_stopped_at_once(monkeypatch):
+    stopped = []
+    monkeypatch.setattr(W, "stop_service", lambda k: stopped.append(k))
+    W._services["longcat:/w/lc#bf16"] = {"proc": None, "port": 1, "url": "u", "meta": {}}
+    MM._measured_vram_gb["longcat:/w/lc#bf16"] = 20.0
+    freed = W.release_vram(5.0)
+    assert stopped == ["longcat:/w/lc#bf16"]
+    assert freed == 20.0                         # the measured figure, not a guess
+
+
+def test_an_in_flight_generation_is_allowed_to_finish(monkeypatch):
+    """The manager cannot see this work: acquire_stt_backend pops the model pool, so
+    _is_key_busy is permanently False and Pass 1 of release_idle_vram would tear the
+    subprocess down mid-generation and lose the whole request."""
+    stopped = []
+    monkeypatch.setattr(W, "stop_service", lambda k: stopped.append(k))
+    monkeypatch.setattr(W, "_drain_timeout_s", lambda: 5.0)
+    key = "longcat:/w/lc#bf16"
+    W._services[key] = {"proc": None, "port": 1, "url": "u", "meta": {}}
+    W._inflight[key] = 1
+    W._idle.clear()
+
+    done = threading.Event()
+
+    def _release():
+        W.release_vram(999.0)
+        done.set()
+
+    t = threading.Thread(target=_release, daemon=True)
+    t.start()
+    assert not done.wait(0.5)                    # waiting, not tearing down
+    assert stopped == []
+    # the generation finishes
+    W._inflight[key] = 0
+    W._idle.set()
+    assert done.wait(5.0)
+    assert stopped == [key]
+
+
+def test_the_drain_has_a_budget(monkeypatch):
+    """A wedged generation must not hold the card for ever."""
+    stopped = []
+    monkeypatch.setattr(W, "stop_service", lambda k: stopped.append(k))
+    monkeypatch.setattr(W, "_drain_timeout_s", lambda: 0.3)
+    key = "longcat:/w/lc#bf16"
+    W._services[key] = {"proc": None, "port": 1, "url": "u", "meta": {}}
+    W._inflight[key] = 1
+    W._idle.clear()
+    assert W.release_vram(999.0) > 0
+    assert stopped == [key]
+
+
+def test_the_drain_budget_is_configurable():
+    assert LongcatConfig().evict_drain_timeout_s == 300.0
+
+
+# ---------------------------------------------------------------- front queue
+def _front(info=None):
+    f = types.SimpleNamespace()
+    f._task_kind = FrontProxy._task_kind
+    f._model_info = lambda m: (info or {})
+    f._queues_at_front = lambda p, m: FrontProxy._queues_at_front(f, p, m)
+    return f
+
+
+def test_text_still_always_queues():
+    assert _front()._queues_at_front("/v1/chat/completions", "lisa")
+
+
+def test_video_queues_only_when_max_instances_is_set():
+    """Queueing video unconditionally would change behaviour for every existing video
+    model and start returning 503 at queue_max_size. Setting the field IS the ask."""
+    assert not _front()._queues_at_front("/v1/video/generations", "lc")
+    assert _front({"max_instances": 2})._queues_at_front("/v1/video/generations", "lc")
+
+
+def test_other_kinds_still_pass_through_unqueued():
+    f = _front({"max_instances": 2})
+    assert not f._queues_at_front("/v1/images/generations", "sd")
+    assert not f._queues_at_front("/v1/embeddings", "bge")
+    assert not f._queues_at_front("/v1/audio/speech", "tts")
+
+
+def test_a_broken_model_lookup_does_not_block_the_request():
+    f = types.SimpleNamespace()
+    f._task_kind = FrontProxy._task_kind
+    f._model_info = lambda m: (_ for _ in ()).throw(RuntimeError("boom"))
+    assert FrontProxy._queues_at_front(f, "/v1/video/generations", "lc") is False
+
+
+def test_every_queue_site_uses_the_helper():
+    """Four sites acquire a slot — direct, keepalive and both brokered. One left on the
+    old text-only guard would make the behaviour depend on which path served you."""
+    src = (ROOT / "codai/frontproxy/app.py").read_text()
+    assert src.count("self._queues_at_front(path, model)") == 4
+    # …and no site is left on the old text-only guard.
+    assert 'self._task_kind(path) == "text"' not in src

tools/longcat_service.py

@@ -263,7 +263,27 @@ def generate(body: dict) -> dict:
         fps = int(body.get("fps") or 30)
     mp4 = _mux(frames, fps)
     return {"mp4_b64": base64.b64encode(mp4).decode(), "num_frames": len(frames),
-            "fps": fps, "stages": list(stages), "segments": 1}
+            "fps": fps, "stages": list(stages), "segments": 1,
+            "vram_gb": _peak_vram_gb()}
+
+
+def _peak_vram_gb() -> float:
+    """The peak VRAM this process reached, in GB, or 0.0.
+
+    Reported back so the engine can size future reservations from a MEASUREMENT instead
+    of an estimate. It has to come from in here: the pipeline loads lazily on the first
+    request, and a host-side free-VRAM delta would also be counting whatever else was
+    loading or evicting on the card at the same time. torch's own peak counter is
+    exactly this process's allocation."""
+    try:
+        import torch
+        if not torch.cuda.is_available():
+            return 0.0
+        # reserved, not allocated: the caching allocator's arenas are what the card is
+        # actually unable to give to anything else.
+        return round(torch.cuda.max_memory_reserved() / (1024 ** 3), 2)
+    except Exception:
+        return 0.0
 
 
 # ── HTTP ──────────────────────────────────────────────────────────────────────