broker/text: move model-load status out of the engine; gate by load_status_updates

The engine-side "Waiting for model reply..." content injection ran inside the
GIL-blocked engine handler during model load, so it never streamed live -- it
only got concatenated into the final reply and polluted the output. Removed it.

Load-status now lives in the responsive front-proxy/broker layer:
- SSE: broker_execute_stream yields a NON-content chunk (empty delta.content +
  x_queue_info status) during each not-ready wait (engine down / starting /
  model loading), so a watching client sees progress without reply pollution.
- out-of-band: the existing broker `pending` keepalives are kept.

Both are gated by a single flag, resolved by _load_status_enabled(model):
- global models.load_status_updates (default True)
- per-model "load_status_updates" override in models.json (unset = inherit)
The same gate drives the SSE side and, via client.status_gate, the out-of-band
`pending` keepalives. Unknown model defaults to on so the relay deadline stays
protected.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RdMufYvtTbtGDWsiZVoXce

codai/api/text.py

@@ -2403,85 +2403,23 @@ async def stream_chat_response(
             # Alternative check for some model managers
             model_loaded = True
     
-    # If model not loaded, add to queue and send waiting notifications
+    # If the model isn't loaded yet, register the request for queue bookkeeping
+    # and just wait. We deliberately do NOT inject any "waiting"/"model starting"
+    # placeholder chunks here: this handler runs inside the engine, whose event
+    # loop is GIL-blocked while the model loads, so anything yielded now never
+    # streams live — it only gets concatenated into the final reply and pollutes
+    # the output. Load-status signalling lives in the front proxy / broker layer
+    # (broker_execute_stream SSE + `pending` keepalives), which stays responsive
+    # during the load and is gated by the load_status_updates flag.
     if not model_loaded:
         await queue_manager.add_waiting(request_id, prefix_key=prefix_key)
-        wait_interval = 2.0  # Send waiting update every 2 seconds
-        last_wait_update = time.time()
-        
-        # Send initial waiting message
-        data = {
-            "id": completion_id,
-            "object": "chat.completion.chunk",
-            "created": created,
-            "model": model_name,
-            "choices": [{
-                "index": 0,
-                "delta": {"content": "Waiting for model reply...\n"},
-                "finish_reason": None,
-            }],
-            "x_queue_info": {
-                "status": "waiting",
-                "message": "Waiting for model reply...",
-            },
-        }
-        yield f"data: {json.dumps(data)}\n\n"
-        
-        # Keep sending wait updates until model is loaded
-        # In a real implementation, this would check a loading status
-        # For now, we'll send a few updates then proceed
-        max_wait_updates = 5
-        wait_count = 0
-        while wait_count < max_wait_updates:
-            await asyncio.sleep(wait_interval)
-            wait_time = await queue_manager.get_wait_time(request_id)
-            wait_count += 1
-            
-            queue_pos = await queue_manager.get_queue_position(request_id)
-            
-            data = {
-                "id": completion_id,
-                "object": "chat.completion.chunk",
-                "created": created,
-                "model": model_name,
-                "choices": [{
-                    "index": 0,
-                    "delta": {"content": f""},
-                    "finish_reason": None,
-                }],
-                "x_queue_info": {
-                    "status": "waiting",
-                    "message": f"Waiting for model reply... ({int(wait_time)}s)",
-                    "queue_position": queue_pos,
-                    "wait_time_seconds": int(wait_time),
-                },
-            }
-            yield f"data: {json.dumps(data)}\n\n"
-    
+
     # Mark as starting processing
     await queue_manager.start_processing(request_id, model_name)
     _tid = task_registry.register("text", title=(model_name or "chat"),
                                   model=model_name or "", task_id=request_id)
     task_registry.start(_tid)
-    
-    # Send "Model starting" message
-    data = {
-        "id": completion_id,
-        "object": "chat.completion.chunk",
-        "created": created,
-        "model": model_name,
-        "choices": [{
-            "index": 0,
-            "delta": {"content": ""},
-            "finish_reason": None,
-        }],
-        "x_queue_info": {
-            "status": "starting",
-            "message": "Model starting",
-        },
-    }
-    yield f"data: {json.dumps(data)}\n\n"
-    
+
     try:
         chunk_count = 0
         _gen_t0 = None          # wall-clock of the first generated token (for it/s)

codai/broker/client.py

@@ -33,6 +33,10 @@ class BrokerClient:
         # Optional async-generator dispatcher for streaming requests: yields SSE
         # chunks which we relay as ``chunk`` envelopes + a terminal ``done``.
         self.stream_dispatcher = None
+        # Optional callback(model) -> bool gating the out-of-band `pending`
+        # keepalives by the per-model / global load_status_updates flag. None or a
+        # True return keeps keepalives on (default), protecting the relay deadline.
+        self.status_gate = None
         self.websocket = None
         self.session_id = None
         self.session_metadata: dict[str, Any] = {}
@@ -407,12 +411,25 @@ class BrokerClient:
             except Exception as _probe_exc:
                 logger.info("CoderAI broker inbound max_tokens probe failed: %s",
                             _probe_exc)
+            # Resolve whether load-status signalling is enabled for this model
+            # (per-model / global load_status_updates, default on). When off, we
+            # suppress the out-of-band `pending` keepalives below — matching the
+            # SSE side in the front proxy. Defaults to on when the gate or model
+            # is unavailable so the relay deadline stays protected.
+            _status_on = True
+            try:
+                _gate = self.status_gate
+                if _gate is not None:
+                    _model_id = _b.get("model") if isinstance(_b, dict) else None
+                    _status_on = bool(_gate(_model_id))
+            except Exception:
+                _status_on = True
             # Send an immediate "pending" acknowledgment so the broker side extends
             # its deadline, then keep sending periodic keepalives while the request
             # runs (e.g. during model download).  The keepalive task is cancelled
             # as soon as we have a final response to send.
             keepalive_task: asyncio.Task | None = None
-            if request_id:
+            if request_id and _status_on:
                 try:
                     await self.websocket.send(json.dumps({
                         "v": 1,

codai/config.py

@@ -100,6 +100,12 @@ class ModelsConfig:
     # clamped to this value. None = no cap (use the client's value, or the 2048
     # fallback). Overridable per-model via the models.json entry's "max_tokens".
     max_tokens: Optional[int] = None
+    # While a model is loading / not ready, signal "still working" so the channel
+    # stays alive and a watching client can show progress: out-of-band broker
+    # `pending` keepalives + a non-content SSE status chunk (no message pollution).
+    # Global default on; override per-model via the models.json entry's
+    # "load_status_updates" (set false to deactivate for that model).
+    load_status_updates: bool = True
 
 
 @dataclass

codai/frontproxy/app.py

@@ -179,6 +179,10 @@ class FrontProxy:
                     query=envelope.query, body=body):
                 yield chunk
         client.stream_dispatcher = _stream_dispatch
+        # Gate the out-of-band `pending` keepalives by the per-model / global
+        # load_status_updates flag (default on). Returns True when the model is
+        # unknown so the relay deadline is still protected by default.
+        client.status_gate = self._load_status_enabled
         self._broker = BrokerService(client)   # app=None → keep our dispatcher
         self._broker.start()
         print("[front] AISBF broker started (front-managed, routes to engines)",
@@ -412,6 +416,26 @@ class FrontProxy:
         # ready, or a non-200 not-ready status).
         _MAX_TRIES = 6
         _RETRY_WAIT = 5.0
+        # Load-status SSE: while we wait for an engine/model to become ready, emit
+        # a NON-content chunk (empty delta.content + x_queue_info) so a watching
+        # client sees "still loading" without polluting the assembled reply. Gated
+        # by the per-model / global load_status_updates flag (default on); the
+        # out-of-band broker `pending` keepalive is gated by the same flag in the
+        # broker client. This runs in the front-proxy event loop, which stays
+        # responsive even while the engine is GIL-blocked loading the model.
+        _status_on = self._load_status_enabled(model)
+        import time as _t0
+
+        def _status_sse(msg):
+            return "data: " + _json.dumps({
+                "id": "chatcmpl-load",
+                "object": "chat.completion.chunk",
+                "created": int(_t0.time()),
+                "model": model or "",
+                "choices": [{"index": 0, "delta": {"content": ""},
+                             "finish_reason": None}],
+                "x_queue_info": {"status": "loading", "message": msg},
+            }) + "\n\n"
 
         def _pick():
             return _router.pick_engine(
@@ -423,6 +447,8 @@ class FrontProxy:
         engine = _pick()
         if engine is None and _is_infer:
             for _ in range(_MAX_TRIES):
+                if _status_on:
+                    yield _status_sse("waiting for an engine to come up")
                 await _asyncio.sleep(_RETRY_WAIT)
                 engine = _pick()
                 if engine is not None:
@@ -475,6 +501,8 @@ class FrontProxy:
                     rp_resp = await self._long.send(rp_req, stream=True)
                 except Exception as exc:
                     if _is_infer and not _last:
+                        if _status_on:
+                            yield _status_sse("engine starting up")
                         await _asyncio.sleep(_RETRY_WAIT)
                         continue
                     yield ('data: {"error":"engine#%s unreachable: %s"}\n\n'
@@ -486,6 +514,8 @@ class FrontProxy:
                 # non-200, so re-sending the body is safe (no duplicate output).
                 if _is_infer and _status in _RETRY_STATUS and not _last:
                     await rp_resp.aclose()
+                    if _status_on:
+                        yield _status_sse("model loading")
                     await _asyncio.sleep(_RETRY_WAIT)
                     continue
                 _meas = (_status == 200 and "text/event-stream"
@@ -559,7 +589,10 @@ class FrontProxy:
                        "engine_fallback": bool(m.get("engine_fallback")),
                        # Per-model concurrency ceiling — the front queue sizes its
                        # gate to this so it never over-subscribes the engine.
-                       "max_instances": m.get("max_instances")}
+                       "max_instances": m.get("max_instances"),
+                       # Per-model override for load-status signalling (None =
+                       # inherit the global models.load_status_updates).
+                       "load_status_updates": m.get("load_status_updates")}
                 for field_ in (m.get("path"), m.get("id"), m.get("alias")):
                     if not field_:
                         continue
@@ -608,6 +641,17 @@ class FrontProxy:
         _models = getattr(self.config, "models", None)
         return max(1, int(getattr(_models, "max_model_instances", 1) or 1))
 
+    def _load_status_enabled(self, model: Optional[str]) -> bool:
+        """Whether to emit load-status signals (out-of-band broker `pending`
+        keepalives + a non-content SSE status chunk) while a model is loading /
+        not ready. Per-model ``load_status_updates`` in models.json wins; else the
+        global ``models.load_status_updates`` (default True)."""
+        ov = self._model_info(model).get("load_status_updates")
+        if ov is not None:
+            return bool(ov)
+        _models = getattr(self.config, "models", None)
+        return bool(getattr(_models, "load_status_updates", True))
+
     def _queue_max_waiting(self) -> int:
         self._refresh_config_if_changed()
         return max(0, int(getattr(self.config.server, "queue_max_size", 6) or 0))