front/engine: responsive Tasks under load — front-side in-flight tracking + thermal cadence

#1 Front-side in-flight task tracking: Engine.enter_request now records per-request
   metadata (model/kind/path/started_at) in engine.active; _merge_engine_tasks
   injects synthetic "running" task entries for in-flight requests not already
   reported by the engine (deduped by engine+model). Both the direct proxy and the
   broker route register/clear it. So the Tasks page shows work the front
   dispatched even when the engine is too GIL-busy generating to answer its own
   /admin/api/tasks. (Combines with the last-good cache.)

#3 Thermal stopping-criteria cadence: the per-token callback held the GIL every
   token, starving admin handlers. Now it does real work every 50 tokens for text,
   scaling to every 100 when generation is fast (>50 tok/s) — far less GIL
   contention, negligible thermal-drift risk between checks.

#2 (no change needed): generation already runs OFF the event loop via
   _aiter_blocking (asyncio.to_thread per token step) for streaming and
   asyncio.to_thread(manager.generate) for non-streaming.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

codai/backends/vulkan.py

@@ -47,15 +47,34 @@ def _make_llama_thermal_criteria():
     llama-cpp-python evaluates stopping criteria synchronously per token inside
     create_(chat_)completion, so blocking here pauses the GPU forward pass —
     mid-generation thermal protection for the GGUF/Vulkan/llama.cpp backend.
-    The criterion never stops generation (returns False) and is throttled so it
-    doesn't read sensors on every token. Returns None if unavailable.
+    The criterion never stops generation (returns False).
+
+    The callback fires PER TOKEN and holds the GIL, which starves the engine's
+    admin handlers (Tasks/status polls) under load. So we only do real work every
+    N tokens: N=50 by default, and N=100 when generation is fast (>50 tok/s), where
+    a per-token GIL grab hurts responsiveness most and thermal drift between checks
+    is negligible. The sensor read itself is still time-throttled in checkpoint().
     """
     try:
         from llama_cpp import StoppingCriteriaList
     except Exception:
         return None
 
+    import time as _time
+    _state = {"n": 0, "interval": 50, "t": _time.monotonic(), "last_n": 0}
+
     def _pause(input_ids, logits):
+        _state["n"] += 1
+        if _state["n"] % _state["interval"] != 0:
+            return False
+        # Re-estimate cadence from the measured token rate over the last window.
+        now = _time.monotonic()
+        dt = now - _state["t"]
+        if dt > 0:
+            rate = (_state["n"] - _state["last_n"]) / dt
+            _state["interval"] = 100 if rate > 50 else 50
+        _state["t"] = now
+        _state["last_n"] = _state["n"]
         try:
             from codai.models.thermal import checkpoint
             checkpoint(context="text-gen", throttle_seconds=2.0)

codai/frontproxy/app.py

@@ -226,6 +226,11 @@ class FrontProxy:
         # Signed with the internal token; the engine accepts it only if it matches.
         if self.internal_token:
             send_headers["x-coderai-broker-authed"] = self.internal_token
+        # Count brokered inference as in-flight too (with metadata) so it shows on
+        # the Tasks page even when the engine is too busy to report it itself.
+        _rid = engine.enter_request(
+            {"model": model or "", "kind": self._task_kind(path), "path": path}
+            if _router.is_inference_path(path) else None)
         try:
             r = await self._long.request(method, engine.url + path,
                                          headers=send_headers, params=query or {},
@@ -237,6 +242,8 @@ class FrontProxy:
                     "headers": {"content-type": "application/json"},
                     "body": ('{"error":"engine#%s unreachable: %s"}'
                              % (engine.id, exc)).encode()}
+        finally:
+            engine.exit_request(_rid)
         # Surface the engine's actual reply so a brokered request that "doesn't get
         # executed" (e.g. an instant small error body) is diagnosable from the log.
         print("[front] broker route: %s %s -> engine#%s(%s) status=%s bytes=%d preview=%r"
@@ -557,8 +564,31 @@ class FrontProxy:
                             "message": e.cooling.get("message")})
         return out
 
+    @staticmethod
+    def _task_kind(path: str) -> str:
+        """Coarse task kind from the inference path, for synthesized in-flight tasks."""
+        p = (path or "").lower()
+        if "/chat/completions" in p or "/completions" in p:
+            return "text"
+        if "/images" in p:
+            return "image"
+        if "/video" in p:
+            return "video"
+        if "/audio/speech" in p or "/tts" in p:
+            return "tts"
+        if "/audio" in p or "/transcri" in p:
+            return "audio"
+        if "/embeddings" in p:
+            return "embedding"
+        return "text"
+
     def _merge_engine_tasks(self, primary, primary_tasks: list) -> list:
-        """Tasks from all engines, each tagged with the engine *name* it runs on."""
+        """Tasks from all engines, each tagged with the engine *name* it runs on.
+
+        Also injects SYNTHETIC entries for requests the front itself has in flight
+        (engine.active) that aren't already represented by a real task — so work
+        shows on the Tasks page even when the target engine is too GIL-busy
+        generating to report it. Deduped by (engine, model)."""
         merged = []
         seen = set()
         # Primary's tasks (from its authed response) — tag with the primary name.
@@ -605,6 +635,32 @@ class FrontProxy:
                 "pausable": False,
                 "restartable": False,
             })
+        # Synthetic in-flight tasks for requests the front dispatched but that have
+        # no real task yet (engine too busy to report). Dedup by (engine, model) so
+        # we don't double-show a generation the engine already reported.
+        import time as _t
+        have = {(t.get("engine"), t.get("model")) for t in merged if isinstance(t, dict)}
+        for e in self.registry.all():
+            for rid, m in list((e.active or {}).items()):
+                key = (e.name, m.get("model") or "")
+                if key in have:
+                    continue
+                have.add(key)
+                merged.append({
+                    "id": f"inflight-{rid}",
+                    "kind": m.get("kind") or "text",
+                    "title": (m.get("model") or "generation"),
+                    "model": m.get("model") or "",
+                    "status": "running",
+                    "step": 0, "total": 0, "rate": 0.0,
+                    "message": "generating",
+                    "started_at": m.get("started_at"),
+                    "engine": e.name,
+                    "active": True,
+                    "cancellable": False,
+                    "pausable": False,
+                    "restartable": False,
+                })
         return merged
 
     # -------------------------------------------------------------------- proxy
@@ -639,11 +695,15 @@ class FrontProxy:
             content=content)
         # Count this as in-flight on the chosen engine so a restart can drain it:
         # decremented only once the response is fully streamed (or send failed).
-        engine.enter_request()
+        # Attach metadata (model/kind) for inference so the front can show a task
+        # for it even when the engine is too busy to answer its own Tasks poll.
+        _meta = ({"model": model or "", "kind": self._task_kind(path), "path": path}
+                 if _router.is_inference_path(path) else None)
+        _rid = engine.enter_request(_meta)
         try:
             rp_resp = await self._long.send(rp_req, stream=True)
         except Exception as exc:
-            engine.exit_request()
+            engine.exit_request(_rid)
             return JSONResponse(
                 {"error": f"Engine#{engine.id} unreachable: {exc}"}, status_code=502)
 
@@ -651,7 +711,7 @@ class FrontProxy:
             try:
                 await rp_resp.aclose()
             finally:
-                engine.exit_request()
+                engine.exit_request(_rid)
 
         resp_headers = self._filter_headers(rp_resp.headers, _DROP_RESP)
         return StreamingResponse(

codai/frontproxy/registry.py

@@ -75,6 +75,10 @@ class Engine:
                                     # and let in-flight ones finish (drain grace period)
     inflight: int = 0              # proxied requests currently streaming through
     _inflight_lock: object = field(default_factory=threading.Lock, repr=False, compare=False)
+    # In-flight request metadata {rid: {"model","kind","path","started_at"}} so the
+    # front can synthesize Tasks-page entries for work it dispatched — visible even
+    # when the engine is too GIL-busy generating to answer its own /admin/api/tasks.
+    active: dict = field(default_factory=dict, repr=False, compare=False)
 
     def __post_init__(self):
         if not self.url:
@@ -105,14 +109,26 @@ class Engine:
         except Exception:
             return True
 
-    def enter_request(self) -> None:
+    def enter_request(self, meta: Optional[dict] = None) -> Optional[str]:
+        """Mark a request in-flight. Returns a request id to pass to exit_request.
+        ``meta`` (model/kind/path) is stored so the front can show a synthetic task
+        for it while the engine itself can't answer the Tasks poll."""
+        import time as _t, uuid as _u
+        rid = _u.uuid4().hex[:12]
         with self._inflight_lock:
             self.inflight += 1
+            if meta is not None:
+                m = dict(meta)
+                m["started_at"] = _t.time()
+                self.active[rid] = m
+        return rid
 
-    def exit_request(self) -> None:
+    def exit_request(self, rid: Optional[str] = None) -> None:
         with self._inflight_lock:
             if self.inflight > 0:
                 self.inflight -= 1
+            if rid:
+                self.active.pop(rid, None)
 
 
 class EngineRegistry: