ds4: per-model launch overrides, multibyte-safe streaming, UI tweaks

Per-model ds4 tuning (these vary by quant/size/context, so they belong on the
model, not globally):
- Optional `ds4` block on a model entry overrides the global Ds4Config for
  ssd_streaming / expert_cache_reserve_gb / extra_args / extra_env; unset fields
  inherit the global config (the default/template). Ds4Backend looks up its own
  model entry and applies the overrides via dataclasses.replace.
- admin: api_model_configure accepts + normalizes the per-model `ds4` block,
  dropping it when empty.
- models page: a "ds4 streaming" section shown only when ds4 is enabled globally
  and the model is a deepseek4; n_ctx stays the context knob.

Fix garbled / truncated ds4 replies: the streaming reader used
iter_lines(decode_unicode=True), which decodes each network chunk independently
and corrupts a multibyte UTF-8 char split across chunks ('—' -> 'â'); the broken
JSON then made json.loads fail and the token was silently dropped (truncated
tails). Parse the SSE byte stream and split on the b"\n" byte (never inside a
UTF-8 sequence), decoding whole lines; also flush a final newline-less line.

UI: slow-reply notice reworded to "Waiting for model reply..." with a trailing
newline so the real reply starts on its own line.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

codai/admin/routes.py

@@ -2686,6 +2686,31 @@ async def api_model_configure(request: Request, username: str = Depends(require_
         if key in data:
             entry[key] = data[key]
 
+    # Per-model ds4 launch overrides (only meaningful for a deepseek4 model served
+    # via the ds4 engine). Normalize to a small dict and drop it when empty, so the
+    # entry stays clean and inherits the global ds4 config as the default.
+    if "ds4" in data:
+        src = data.get("ds4") if isinstance(data.get("ds4"), dict) else {}
+        ds4o = {}
+        sv = src.get("ssd_streaming")
+        if sv is not None and sv != "":
+            ds4o["ssd_streaming"] = (sv.lower() in ("1", "true", "on", "yes")
+                                     if isinstance(sv, str) else bool(sv))
+        rg = src.get("expert_cache_reserve_gb")
+        if rg not in (None, "", 0, "0"):
+            try:
+                ds4o["expert_cache_reserve_gb"] = max(0, int(rg))
+            except (TypeError, ValueError):
+                pass
+        for k in ("extra_args", "extra_env"):
+            v = src.get(k)
+            if isinstance(v, str) and v.strip():
+                ds4o[k] = v.strip()
+        if ds4o:
+            entry["ds4"] = ds4o
+        else:
+            entry.pop("ds4", None)
+
     # A GGUF LLM is served by llama.cpp. Its multimodal projector (mmproj) gives
     # it VISION INPUT, which is the `image_to_text` capability served through
     # llama.cpp — NOT the diffusers `vision_models`/`image_models` categories

codai/admin/templates/models.html

@@ -573,6 +573,36 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
         </label>
         <span class="form-hint" style="font-size:11px">When off (default), a request fails if the pinned engine is down/busy-unreachable instead of running on a different card.</span>
       </div>
+      <div class="form-row" id="cfg-ds4-row" style="margin-top:.75rem;display:none;border:1px solid var(--border,#333);border-radius:8px;padding:.6rem">
+        <label class="form-label" style="font-weight:600">ds4 (DeepSeek V4) streaming <span class="muted" style="font-weight:400">— per-model overrides</span></label>
+        <span class="form-hint" style="font-size:11px;margin-bottom:.4rem">These tune ds4-server for THIS model (they vary by quant/size/context). Leave blank to inherit the global ds4 settings. Context window is set by <b>n_ctx</b> above.</span>
+        <div style="display:grid;grid-template-columns:1fr 1fr;gap:.6rem">
+          <div class="form-row" style="margin:0">
+            <label class="form-label">SSD expert streaming</label>
+            <select id="cfg-ds4-ssd" class="form-input">
+              <option value="">Inherit global</option>
+              <option value="true">On — stream experts from disk</option>
+              <option value="false">Off — full residency</option>
+            </select>
+            <span class="form-hint" style="font-size:11px">On for big models that don't fit VRAM; off for small ones that do.</span>
+          </div>
+          <div class="form-row" style="margin:0">
+            <label class="form-label">Expert-cache VRAM reserve (GiB)</label>
+            <input type="number" id="cfg-ds4-reserve" class="form-input" min="0" placeholder="0 = inherit / ds4 default">
+            <span class="form-hint" style="font-size:11px">Set just above this model's resident weights (+~2 GiB). ds4 defaults to half the card, which starves the cache.</span>
+          </div>
+        </div>
+        <div class="form-row" style="margin:.5rem 0 0">
+          <label class="form-label">Extra ds4-server args</label>
+          <input type="text" id="cfg-ds4-extra-args" class="form-input" placeholder="--ssd-streaming-cache-experts 300">
+          <span class="form-hint" style="font-size:11px">Cap the expert cache (a count or NGB), prefill-chunk, kv-disk-space, etc. Blank = inherit global.</span>
+        </div>
+        <div class="form-row" style="margin:.5rem 0 0">
+          <label class="form-label">Extra ds4-server env</label>
+          <input type="text" id="cfg-ds4-extra-env" class="form-input" placeholder="DS4_CUDA_WEIGHT_ARENA_CHUNK_MB=512">
+          <span class="form-hint" style="font-size:11px"><code>KEY=VALUE</code> pairs. e.g. <code>DS4_CUDA_WEIGHT_ARENA_CHUNK_MB=512</code> fixes weight-arena OOM under heavy expert caching. Blank = inherit global.</span>
+        </div>
+      </div>
       <div style="display:grid;grid-template-columns:1fr 1fr;gap:.75rem;margin-top:.75rem">
         <div class="form-row" style="margin:0">
           <label class="form-label">Used VRAM <span class="muted">(GB)</span></label>
@@ -1039,6 +1069,7 @@ function closeModal(id){document.getElementById(id).classList.remove('show')}
 /* ── Global settings ─────────────────────────────────── */
 let _defaultOffloadDir = './offload';
 let _highlightCap = null;  // capability to highlight in local models list (from ?local_cap= param)
+let _ds4Enabled = false;   // whether the ds4 (DeepSeek V4) engine is enabled globally
 
 async function loadGlobalSettings(){
   try{
@@ -1046,6 +1077,7 @@ async function loadGlobalSettings(){
     if(r.ok){
       const d = await r.json();
       _defaultOffloadDir = d.offload?.directory || './offload';
+      _ds4Enabled = !!(d.ds4 && d.ds4.enabled);
     }
   }catch{}
 }
@@ -3173,6 +3205,7 @@ function openCfgModal(idx, cfgIdx){
   _applyBackendCfgVisibility(m);
   _populateEnginePin(s.engine || '');
   document.getElementById('cfg-engine-fallback').checked = !!s.engine_fallback;
+  _populateDs4(m, s);
   document.getElementById('cfg-sysprompt').value = s.system_prompt || '';
   document.getElementById('cfg-parser').value = s.parser || (!m.in_config ? _autoDetectParser(m.path) : 'auto');
   document.getElementById('cfg-tools').checked = !!s.tools_closer_prompt;
@@ -3484,6 +3517,39 @@ async function _populateEnginePin(desired){
   } catch(e) { row.style.display = 'none'; }
 }
 
+// Per-model ds4 overrides: shown only when the ds4 engine is enabled globally and
+// the model is a deepseek4 (the engine that ds4-server serves). Fields left blank
+// inherit the global ds4 config.
+function _populateDs4(m, s){
+  const row = document.getElementById('cfg-ds4-row');
+  if(!row) return;
+  const hay = (((m && (m.path||m.id)) || '') + ' ' + ((s && s.alias) || '')).toLowerCase();
+  const isDeepSeek4 = /deepseek[\s\-_]?v4|ds4/.test(hay);
+  row.style.display = (_ds4Enabled && isDeepSeek4) ? '' : 'none';
+  const d = (s && s.ds4) || {};
+  document.getElementById('cfg-ds4-ssd').value =
+    (d.ssd_streaming === true ? 'true' : (d.ssd_streaming === false ? 'false' : ''));
+  document.getElementById('cfg-ds4-reserve').value =
+    (d.expert_cache_reserve_gb != null ? d.expert_cache_reserve_gb : '');
+  document.getElementById('cfg-ds4-extra-args').value = d.extra_args || '';
+  document.getElementById('cfg-ds4-extra-env').value = d.extra_env || '';
+}
+
+// Collect the per-model ds4 overrides; only set fields are included (blank =
+// inherit). The backend normalizes and drops the block when empty.
+function _collectDs4(){
+  const ssd = document.getElementById('cfg-ds4-ssd').value;       // '' | 'true' | 'false'
+  const reserve = parseInt(document.getElementById('cfg-ds4-reserve').value);
+  const ea = document.getElementById('cfg-ds4-extra-args').value.trim();
+  const ev = document.getElementById('cfg-ds4-extra-env').value.trim();
+  const o = {};
+  if(ssd === 'true' || ssd === 'false') o.ssd_streaming = (ssd === 'true');
+  if(!isNaN(reserve) && reserve > 0) o.expert_cache_reserve_gb = reserve;
+  if(ea) o.extra_args = ea;
+  if(ev) o.extra_env = ev;
+  return o;
+}
+
 async function saveModelConfig(){
   const path = document.getElementById('cfg-path').value;
   const maxGpu = parseFloat(document.getElementById('cfg-max-gpu').value);
@@ -3527,6 +3593,7 @@ async function saveModelConfig(){
     offload_dir:       document.getElementById('cfg-offload-dir').value.trim() || './offload',
     engine:            document.getElementById('cfg-engine').value.trim() || null,
     engine_fallback:   document.getElementById('cfg-engine-fallback').checked,
+    ds4:               _collectDs4(),
     system_prompt:     document.getElementById('cfg-sysprompt').value.trim() || null,
     parser:            document.getElementById('cfg-parser').value,
     tools_closer_prompt: document.getElementById('cfg-tools').checked,

codai/api/text.py

@@ -1789,7 +1789,7 @@ async def stream_chat_response(
             "model": model_name,
             "choices": [{
                 "index": 0,
-                "delta": {"content": "Waiting for model reply..."},
+                "delta": {"content": "Waiting for model reply...\n"},
                 "finish_reason": None,
             }],
             "x_queue_info": {

codai/backends/ds4.py

@@ -61,11 +61,76 @@ class Ds4Backend(ModelBackend):
         # Resolve the requested model to a concrete .gguf so ds4-server serves THE
         # deepseek4 model that was asked for (not a fixed downloaded variant).
         model_file = self._resolve_gguf(model_name)
+        # Per-model ds4 launch overrides (streaming / expert-cache reserve / extra
+        # args+env) from THIS model's own config entry — these vary by quant and
+        # context, so they live per-model. Unset fields inherit the global ds4
+        # config, which acts as the default/template.
+        overrides = self._ds4_overrides(model_name, model_file)
+        if overrides:
+            import dataclasses
+            try:
+                self._cfg = dataclasses.replace(self._cfg, **overrides)
+                print(f"[ds4] per-model overrides for '{model_name}': "
+                      + ", ".join(f"{k}={v!r}" for k, v in overrides.items()), flush=True)
+            except Exception as exc:
+                print(f"[ds4] failed to apply per-model overrides: {exc}", flush=True)
         _resolved, self._svc_key = ds4_worker.resolve_service_key(self._cfg, model_file)
         self._url = ds4_worker.ensure_service(
             self._cfg, model_file=model_file, ctx=(self._ctx or None),
             model_name=model_name)
 
+    @staticmethod
+    def _ds4_overrides(model_name: str, model_file: Optional[str]) -> Dict:
+        """Per-model ds4 overrides from the model's own models.json entry.
+
+        Looks up THIS model's config entry (by resolved path or name) and returns
+        the subset of :class:`Ds4Config` fields its optional ``ds4`` block sets:
+        ``ssd_streaming``, ``expert_cache_reserve_gb``, ``extra_args``,
+        ``extra_env``. Unset/blank fields are omitted so the global ds4 config
+        stays the default. Best-effort: any failure yields no overrides."""
+        out: Dict = {}
+        try:
+            import os
+            from codai.admin.routes import config_manager
+            md = getattr(config_manager, "models_data", {}) or {}
+            target = os.path.basename(os.path.expanduser(model_file or "")) or None
+            name_l = (model_name or "").strip().lower()
+            entry = None
+            for lst in md.values():
+                if not isinstance(lst, list):
+                    continue
+                for m in lst:
+                    if not isinstance(m, dict):
+                        continue
+                    path = str(m.get("path") or m.get("id") or "")
+                    base = os.path.basename(path)
+                    stem = base[:-5] if base.lower().endswith(".gguf") else base
+                    cands = {path.lower(), base.lower(), stem.lower(),
+                             str(m.get("alias") or "").lower()}
+                    if (target and base == target) or (name_l and name_l in cands):
+                        entry = m
+                        break
+                if entry:
+                    break
+            ds4o = entry.get("ds4") if entry and isinstance(entry.get("ds4"), dict) else None
+            if not ds4o:
+                return out
+            if ds4o.get("ssd_streaming") is not None:
+                out["ssd_streaming"] = bool(ds4o["ssd_streaming"])
+            rg = ds4o.get("expert_cache_reserve_gb")
+            if rg not in (None, "", 0, "0"):
+                try:
+                    out["expert_cache_reserve_gb"] = max(0, int(rg))
+                except (TypeError, ValueError):
+                    pass
+            for k in ("extra_args", "extra_env"):
+                v = ds4o.get(k)
+                if v and str(v).strip():
+                    out[k] = str(v).strip()
+        except Exception:
+            pass
+        return out
+
     @staticmethod
     def _resolve_gguf(model_name: str):
         """Map a requested model name/alias/path/basename to a local .gguf path.
@@ -187,29 +252,53 @@ class Ds4Backend(ModelBackend):
         queue: asyncio.Queue = asyncio.Queue()
         _SENTINEL = object()
 
+        def _handle_line(line: str) -> bool:
+            """Process one SSE line; return True when the stream should stop."""
+            if not line or not line.startswith("data:"):
+                return False
+            data = line[len("data:"):].strip()
+            if data == "[DONE]":
+                return True
+            try:
+                obj = json.loads(data)
+            except ValueError:
+                return False
+            choice = (obj.get("choices") or [{}])[0]
+            text = (choice.get(delta_key) or {}).get("content") or ""
+            if text:
+                loop.call_soon_threadsafe(queue.put_nowait, text)
+            if obj.get("usage"):
+                self._store_usage(obj["usage"])
+            return bool(choice.get("finish_reason"))
+
         def _worker():
             import requests
             try:
                 with requests.post(url, json=payload, stream=True, timeout=3600) as r:
                     r.raise_for_status()
-                    for raw in r.iter_lines(decode_unicode=True):
-                        if not raw or not raw.startswith("data:"):
-                            continue
-                        data = raw[len("data:"):].strip()
-                        if data == "[DONE]":
-                            break
-                        try:
-                            obj = json.loads(data)
-                        except ValueError:
+                    # Parse the SSE byte stream ourselves and split on the b"\n"
+                    # byte. requests' iter_lines(decode_unicode=True) decodes each
+                    # network chunk independently, so a multibyte UTF-8 char split
+                    # across chunks gets corrupted — which breaks that event's JSON
+                    # and silently drops the token (mangled '—', truncated tails).
+                    # b"\n" (0x0A) can never fall inside a UTF-8 multibyte sequence,
+                    # so splitting bytes first and decoding whole lines is safe.
+                    buf = b""
+                    done = False
+                    for bchunk in r.iter_content(chunk_size=8192):
+                        if not bchunk:
                             continue
-                        choice = (obj.get("choices") or [{}])[0]
-                        text = (choice.get(delta_key) or {}).get("content") or ""
-                        if text:
-                            loop.call_soon_threadsafe(queue.put_nowait, text)
-                        if obj.get("usage"):
-                            self._store_usage(obj["usage"])
-                        if choice.get("finish_reason"):
+                        buf += bchunk
+                        while b"\n" in buf:
+                            raw, buf = buf.split(b"\n", 1)
+                            if _handle_line(raw.decode("utf-8", "replace").strip()):
+                                done = True
+                                break
+                        if done:
                             break
+                    # Flush any final line the stream left without a trailing newline.
+                    if not done and buf.strip():
+                        _handle_line(buf.decode("utf-8", "replace").strip())
             except Exception as exc:  # surface to the consumer
                 loop.call_soon_threadsafe(queue.put_nowait, exc)
             finally: