Upload app_deepseek.py

app_deepseek.py

@@ -1,6 +1,6 @@
 """
 gijl/agentic-thinker-v1 -- multi-loop dataset generator (single-file app.py)
-v8, worker "deepseek" (DeepSeek-V4-Flash-0731): llama.cpp backend + multi-loop + verified, source-grouped, auditable
+v8, worker "deepseek" (DeepSeek-V4-Flash-0731): ds4 backend (v8.2; was llama.cpp) + multi-loop + verified, source-grouped, auditable
     output, + genuine agent-task trajectories + five specialized workers (glm/minimax/deepseek/kimi/
     qwen) concurrent collection, now with real per-model output-compatibility handling. Sibling files
     run the identical pipeline -- see WHAT CHANGED IN v7 and WHAT CHANGED IN v8.
@@ -310,17 +310,27 @@ architecture to mature (GIJL_LLAMA_SERVER is unset by default, so a later run au
 whatever "latest" is by then -- no file change needed), or point GIJL_LLAMA_SERVER (already supported,
 see ensure_llama_server()) at a self-built binary from an experimental third-party CPU-capable fork --
 unofficial, so verify it yourself before trusting it with real data.
+v8.2 (backend swap, this worker only): llama.cpp is REMOVED from this file and replaced by antirez/ds4 (DwarfStar),
+the inference engine written specifically for DeepSeek-V4-Flash -- section 5b. `make cpu` builds a CPU-only ds4-server
+from source on first start (plain C, a few minutes, built in GIJL_DS4_BUILD_DIR, not synced to the hub because it is
+-march=native). ds4 loads ONLY its own GGUFs: GGUF_REPO is now antirez/deepseek-v4-gguf and GGUF_QUANT is one of
+q2 (default, ~81GiB) | q2-q4 (~98GB) | q4 -- the unsloth/llama.cpp GGUFs no longer apply. ds4 renders the DeepSeek chat
+template itself, so the client sends plain `messages` to /v1/chat/completions (tool role folded into user turns, `tools`
+not forwarded, thinking off via `think:false` from _CHAT_TEMPLATE_EXTRA_KWARGS) instead of a locally-rendered raw
+prompt; the tokenizer is no longer loaded for inference. NOTHING outside section 5b, the backend constants, the two
+`bin/*` sync/restore patterns and _start_backend() was touched. UNVERIFIED in this file (no 90GB box available when it
+was written): real CPU speed, and whether ds4-server cancels generation when the client disconnects -- run
+GIJL_SMOKE=1 first and read the [ds4] lines.
 FOUR sibling files (glm/minimax/deepseek/kimi/qwen minus this one) run the same pipeline, each specialized
 differently -- see WHAT CHANGED IN v7 and WHAT CHANGED IN v8 at the top of this file for the full table and the
-evidence behind each choice. all five write into the SAME dataset concurrently -- see MODEL_TAG and the v6 note above for how that stays race-free. The chat template + tokenizer come from the
-MODEL_ID repo (rendered locally), the weights come from the GGUF repo. ONE llama-server process, ONE generation
+evidence behind each choice. all five write into the SAME dataset concurrently -- see MODEL_TAG and the v6 note above for how that stays race-free. (v8.2: ds4 renders the chat template itself; the weights come from the GGUF repo.) ONE ds4-server process, ONE generation
 stream (parallel slots = 1), context 12288 tokens by default (GIJL_CTX). The tool-call XML convention below
 (<atem:...>) is taught IN THE PROMPT (see build_scout_prompt/build_agent_task_prompt), not a trained-in format of
 any one model, so it is unchanged across workers -- but it is still worth a GIJL_SMOKE=1 run per worker before a
 long session, to confirm this model actually follows it turn after turn.
 
-HARDWARE (llama.cpp backend): NVIDIA GPU via a CUDA build, otherwise CPU. llama-server's --fit (default) fits layers
-to the free VRAM; when the whole model fits, every layer is offloaded (-ngl 999). TPU is not supported by llama.cpp.
+HARDWARE (ds4 backend, v8.2): CPU by default (`make cpu`, 16 vCPUs / ~103GB file ceiling as in v7); GIJL_DS4_BUILD=cuda-generic
+builds ds4's CUDA path instead. TPU is not supported.
 (The old transformers backend's TPU notes: device_map="auto" did NOT place the model on the TPU -- the log showed
 "no hf_device_map" -- and generation stalled in XLA compilation.)
 
@@ -366,8 +376,8 @@ Before running:
    Stack Exchange key (label must match SECRET_NAMES below).
 3. Optional env: GIJL_ONLY_LOOPS=judgment_dpo,sft_helpful  (restrict loops),
    GIJL_TARGET_REPO=user/repo, GIJL_WORK_DIR=/some/dir, GIJL_MODEL_ID=..., GIJL_CTX=12288,
-   GIJL_GGUF_QUANT=UD-Q4_K_XL, GIJL_LLAMA_SERVER=/path/to/your/llama-server, GIJL_REBUILD_LLAMA=1,
-   GIJL_CACHE_LLAMA_BINARY=0, GIJL_ALLOW_CPU_FALLBACK=1, GIJL_REASONING_GEN=low, GIJL_REASONING_SCOUT=medium
+   GIJL_GGUF_QUANT=q2|q2-q4|q4, GIJL_DS4_SERVER=/path/to/your/ds4-server, GIJL_REBUILD_DS4=1,
+   GIJL_DS4_REF=<tag/branch to pin>, GIJL_DS4_BUILD=cpu|cuda-generic, GIJL_REASONING_GEN=low, GIJL_REASONING_SCOUT=medium
 """
 
 import os
@@ -539,21 +549,13 @@ TARGET_DATASET_REPO = os.environ.get("GIJL_TARGET_REPO", "gijl/agentic-thinker-v
 SEARCH_LANGUAGE_CYCLE = ["arabic", "english"]
 
 # --- inference backend -------------------------------------------------------
-# "llamacpp" (default): llama-server + GGUF, on GPU (CUDA build) or CPU. llama.cpp cannot use TPU.
+# "ds4" (default): antirez/ds4 (DwarfStar) ds4-server + its own GGUF, built from source with `make cpu` (CPU) -- the
+# engine written specifically for DeepSeek-V4-Flash; llama.cpp is no longer used. See section 5b.
 # "hf": the old transformers path (only kept for reference; on TPU it produced ~3 chars/s and hung).
-BACKEND = os.environ.get("GIJL_BACKEND", "llamacpp").strip().lower()
-GGUF_REPO = os.environ.get("GIJL_GGUF_REPO", "unsloth/DeepSeek-V4-Flash-0731-GGUF")
-GGUF_QUANT = os.environ.get("GIJL_GGUF_QUANT", "UD-Q2_K_XL")   # v7: pinned explicitly -- FIXED from v6's generic "Q6_K" default, which is not a real tag in this repo's quant ladder -- confirmed 96.8GB (unsloth's published quant-measurement table); UD-IQ3_XXS, the next step up, is 104.2GB and does not fit
-                                                           # choose_quant()/_memory_budget_gb() still run underneath this as a
-                                                           # safety net (step down further on a smaller box); GIJL_GGUF_QUANT
-                                                           # overrides this default at launch time.
-LLAMA_CTX = int(os.environ.get("GIJL_CTX", "12288"))            # tokens; KV cache is ~2-4GB at this size
-LLAMA_LOAD_TIMEOUT_S = 1800
-LLAMA_BIN_DIR = os.path.join(WORK_DIR, "bin")                   # CUDA build cache (synced to the hub, see CACHE_LLAMA_BINARY)
-LLAMA_CPU_DIR = os.path.join(WORK_DIR, "llama_cpu")             # prebuilt CPU release (not synced: quick to re-download)
-LLAMA_BUILD_ROOT = os.environ.get("GIJL_LLAMA_BUILD_DIR", "/tmp/gijl_llama_build")
-CACHE_LLAMA_BINARY = os.environ.get("GIJL_CACHE_LLAMA_BINARY", "1") != "0"
-LLAMA_BUILD_TIMEOUT_S = 45 * 60
+BACKEND = os.environ.get("GIJL_BACKEND", "ds4").strip().lower()
+GGUF_REPO = os.environ.get("GIJL_GGUF_REPO", "antirez/deepseek-v4-gguf")   # ds4's own GGUF repo (the unsloth/llama.cpp GGUFs do NOT load in ds4)
+GGUF_QUANT = os.environ.get("GIJL_GGUF_QUANT", "q2")   # q2 (~81GiB, ds4f-q2, default) | q2-q4 (~98GB) | q4 (~153GiB) -- see DS4_GGUF_FILES; q2-q4 also fits the 103GB ceiling but needs more RAM
+DS4_CTX = int(os.environ.get("GIJL_CTX", "12288"))              # tokens passed to ds4-server --ctx
 # Meta's recommended sampling is temperature 1.0 / top_p 0.95 / top_k 64. A slightly lower temperature
 # suits structured-JSON tasks; top_p/top_k stay as recommended.
 TOP_P, TOP_K = 0.95, 64
@@ -1326,12 +1328,9 @@ class ModelClient:
 
 
 # ---------------------------------------------------------------------------
-# 5b. llama.cpp backend (DEFAULT): CUDA build on GPU, prebuilt binary on CPU.
-#     llama.cpp cannot use a TPU. Behavior below was checked against a real
-#     llama-server (build b10964) running a tiny GGUF: /completion SSE format,
-#     /tokenize double-BOS, HTTP 400 on context overflow, cancel-on-disconnect.
-#     There are NO official Linux CUDA binaries (only Windows CUDA), so the GPU
-#     path builds from source (cached in the hub repo under bin/).
+# 5b. ds4 (DwarfStar) backend (DEFAULT): a DeepSeek-V4-Flash-specific engine, CPU build (`make cpu`) or GPU build.
+#     Replaces the llama.cpp backend (llama-server crashed on this model, see the v8.1 field report above).
+#     ds4 only runs its own GGUFs (antirez/deepseek-v4-gguf) and renders the chat template itself.
 # ---------------------------------------------------------------------------
 def _run(cmd, timeout=60, env=None):
     try:
@@ -1399,279 +1398,94 @@ def _memory_budget_gb(acc: dict) -> float:
     return (acc["ram_gb"] or 24.0) * 0.75 - 2.0
 
 
-# --- GGUF selection / download ---------------------------------------------------
-_QUANT_RE = re.compile(r"-((?:UD-)?(?:IQ|Q)\d[A-Za-z0-9_]*|BF16|F16|F32|MXFP4[A-Za-z0-9_]*|NVFP4[A-Za-z0-9_]*)$", re.IGNORECASE)
-_SHARD_RE = re.compile(r"-\d{5}-of-\d{5}$")
-
-
-def _group_gguf_files(siblings) -> dict:
-    """siblings: iterable of (filename, size_bytes) -> {QUANT_TAG: {"files": [...], "size_gb": float}}.
-    Skips vision projectors and speculative-decoding drafters; joins split shards under one tag."""
-    out = {}
-    for name, size in siblings:
-        base = os.path.basename(name)
-        low = base.lower()
-        if not low.endswith(".gguf") or any(x in low for x in ("mmproj", "dflash", "assistant", "draft")):
-            continue
-        m = _QUANT_RE.search(_SHARD_RE.sub("", base[:-5]))
-        if not m:
-            continue
-        g = out.setdefault(m.group(1).upper(), {"files": [], "size_gb": 0.0})
-        g["files"].append(name)
-        g["size_gb"] += (size or 0) / 1e9
-    for g in out.values():
-        g["files"].sort()
-    return out
-
-
-def choose_quant(groups: dict, budget_gb: float, want: str, explicit: bool):
-    """-> (tag, reason). An explicitly requested quant is always honored; the default one steps down if it
-    does not fit the memory budget."""
-    if not groups:
-        raise RuntimeError(f"no usable .gguf files found in {GGUF_REPO}")
-    want_u = (want or "").upper()
-    if want_u in groups and (explicit or groups[want_u]["size_gb"] <= budget_gb):
-        return want_u, "requested"
-    fitting = [t for t, g in groups.items() if g["size_gb"] <= budget_gb]
-    if want_u in groups:
-        smaller = [t for t in fitting if groups[t]["size_gb"] < groups[want_u]["size_gb"]]
-        if smaller:
-            return max(smaller, key=lambda t: groups[t]["size_gb"]), f"{want_u} ({groups[want_u]['size_gb']:.1f}GB) exceeds the {budget_gb:.1f}GB budget"
-    elif fitting:
-        return max(fitting, key=lambda t: groups[t]["size_gb"]), f"{want_u or 'requested quant'} not in the repo"
-    return min(groups, key=lambda t: groups[t]["size_gb"]), "nothing fits the budget: smallest quant, expect partial offload"
+# --- ds4 (DwarfStar) backend: the DeepSeek-V4-Flash-specific engine, replaces llama.cpp ------------------------
+# WHY (field report in the module docstring, v8.1): llama-server crashed on CPU at startup with this model
+# (GGML_ASSERT(*cur_backend_id != -1) in ggml_backend_sched_split_graph) -- llama.cpp's DeepSeek-V4 support is
+# still the weak spot, not this script. antirez/ds4 is an inference engine written ONLY for DeepSeek V4 Flash:
+#   * `make cpu` builds a CPU-only ds4-server (Linux, plain C, no CUDA/Metal needed) -- confirmed in its Makefile.
+#   * it runs ONLY its own GGUF files (repo antirez/deepseek-v4-gguf, "-0731" files = Flash 0731), NOT the
+#     unsloth/llama.cpp GGUFs. GIJL_GGUF_REPO/GIJL_GGUF_QUANT therefore now name THAT repo and one of the
+#     DS4_GGUF_FILES keys below (q2 ~81GiB | q2-q4 ~98GB | q4 ~153GiB).
+#   * it renders the DeepSeek chat template itself (incl. thinking on/off), so this client sends plain chat
+#     `messages` to POST /v1/chat/completions instead of a locally-rendered raw prompt to /completion. The tokenizer
+#     is therefore not needed for inference any more. `tools` is deliberately NOT forwarded: the atem: tool protocol
+#     is taught in the system prompt (see section 6), and a native `tools` field would make the server parse
+#     DeepSeek's own DSML tool calls into a structured field this pipeline never reads.
+DS4_REPO_URL = os.environ.get("GIJL_DS4_REPO_URL", "https://github.com/antirez/ds4")
+DS4_REF = os.environ.get("GIJL_DS4_REF", "").strip()              # optional tag/branch/commit to pin (main moves fast)
+DS4_BUILD_TARGET = os.environ.get("GIJL_DS4_BUILD", "cpu").strip()  # `make <target>`; anything but "cpu" is a GPU build
+DS4_BUILD_ROOT = os.environ.get("GIJL_DS4_BUILD_DIR", "/tmp/gijl_ds4_build")
+DS4_BUILD_TIMEOUT_S = 30 * 60
+DS4_LOAD_TIMEOUT_S = 3600                                         # ~80-100GB GGUF has to be read from disk first
+DS4_MODEL_ALIAS = "deepseek-v4-flash"                             # the server ignores it (the GGUF picks the model)
+DS4_GGUF_FILES = {   # exact names from ds4's own download_model.sh (ds4f-q2 / ds4f-q2-q4 / ds4f-q4)
+    "q2": "DeepSeek-V4-Flash-IQ2XXS-w2Q2K-AProjQ8-SExpQ8-OutQ8-chat-v2-imatrix-0731.gguf",
+    "q2-q4": "DeepSeek-V4-Flash-Layers37-42Q4KExperts-OtherExpertLayersIQ2XXSGateUp-Q2KDown-AProjQ8-SExpQ8-OutQ8-chat-v2-imatrix-fixed-0731.gguf",
+    "q4": "DeepSeek-V4-Flash-Q4KExperts-F16HC-F16Compressor-F16Indexer-Q8Attn-Q8Shared-Q8Out-chat-v2-imatrix-0731.gguf",
+}
+DS4_GGUF_SIZE_GB = {"q2": 87.0, "q2-q4": 98.0, "q4": 164.0}      # approximate, from ds4's own download notes
 
 
 def download_gguf(acc: dict):
-    """-> (path_to_first_shard, {"quant","size_gb"}). Stored in the default HF cache (NOT /kaggle/working,
-    which is capped at ~20GB of saved output)."""
-    from huggingface_hub import HfApi, hf_hub_download
+    """-> (path_to_gguf, {"quant","size_gb"}). Stored in the default HF cache (NOT /kaggle/working)."""
+    from huggingface_hub import hf_hub_download
     token = os.environ.get("HF_TOKEN")
     explicit_file = os.environ.get("GIJL_GGUF_FILE")
     if explicit_file:
         path = hf_hub_download(GGUF_REPO, explicit_file, token=token)
         return path, {"quant": explicit_file, "size_gb": os.path.getsize(path) / 1e9}
-    try:
-        info = HfApi(token=token).model_info(GGUF_REPO, files_metadata=True)
-        groups = _group_gguf_files((s.rfilename, getattr(s, "size", 0)) for s in info.siblings)
-    except Exception as e:  # noqa: BLE001 -- fall back to the naming scheme from the model docs
-        # v6: GGUF_REPO now differs per worker (glm/minimax/deepseek) -- guess the base name from the repo id
-        # itself (unsloth's own convention: "<org>/<Model-Name>-GGUF" -> "<Model-Name>-<QUANT>.gguf") instead of
-        # a name hardcoded for one specific model.
-        guess = GGUF_REPO.rsplit("/", 1)[-1].removesuffix("-GGUF")
-        print(f"[llama] could not list {GGUF_REPO} ({_redact(str(e))[:120]}); assuming {guess}-{GGUF_QUANT}.gguf", flush=True)
-        groups = {GGUF_QUANT.upper(): {"files": [f"{guess}-{GGUF_QUANT}.gguf"], "size_gb": 0.0}}
-    budget = _memory_budget_gb(acc)
-    tag, why = choose_quant(groups, budget, GGUF_QUANT, "GIJL_GGUF_QUANT" in os.environ)
-    files = groups[tag]["files"]
-    print(f"[llama] model: {GGUF_REPO} quant={tag} ({groups[tag]['size_gb']:.1f}GB, budget {budget:.1f}GB) -- {why}", flush=True)
+    tag = re.sub(r"^ds4f-", "", (GGUF_QUANT or "q2").strip().lower())
+    if tag not in DS4_GGUF_FILES:
+        raise RuntimeError(f"GIJL_GGUF_QUANT={GGUF_QUANT!r} is not a ds4 quant; use one of {sorted(DS4_GGUF_FILES)} "
+                           f"(or GIJL_GGUF_FILE=<exact file name inside {GGUF_REPO}>)")
+    size_gb = DS4_GGUF_SIZE_GB[tag]
+    ram = acc.get("ram_gb") or 0.0
+    if ram and size_gb > ram * 0.92:
+        print(f"[ds4][WARNING] quant={tag} (~{size_gb:.0f}GB) is close to / above this machine's RAM ({ram:.0f}GB): "
+              f"expect swapping or a failed load. Try GIJL_GGUF_QUANT=q2 (~87GB).", flush=True)
+    print(f"[ds4] model: {GGUF_REPO}/{DS4_GGUF_FILES[tag]} quant={tag} (~{size_gb:.0f}GB, RAM {ram:.0f}GB)", flush=True)
     t0 = time.time()
-    paths = [hf_hub_download(GGUF_REPO, f, token=token) for f in files]
-    print(f"[llama] GGUF ready in {time.time() - t0:.0f}s: {paths[0]}", flush=True)
-    return paths[0], {"quant": tag, "size_gb": groups[tag]["size_gb"] or os.path.getsize(paths[0]) / 1e9}
-
-
-# --- getting a llama-server binary ---------------------------------------------------
-def _llama_env(lib_dir):
-    env = os.environ.copy()
-    if lib_dir:
-        env["LD_LIBRARY_PATH"] = lib_dir + (":" + env["LD_LIBRARY_PATH"] if env.get("LD_LIBRARY_PATH") else "")
-    return env
+    path = hf_hub_download(GGUF_REPO, DS4_GGUF_FILES[tag], token=token)
+    print(f"[ds4] GGUF ready in {time.time() - t0:.0f}s: {path}", flush=True)
+    return path, {"quant": tag, "size_gb": os.path.getsize(path) / 1e9}
 
 
-def _binary_ok(binary: str, lib_dir, need_cuda: bool) -> bool:
-    env = _llama_env(lib_dir)
-    rc, out, err = _run([binary, "--version"], 90, env)
-    if rc != 0:
-        return False
-    if need_cuda:
-        rc, out, err = _run([binary, "--list-devices"], 90, env)
-        return rc == 0 and "cuda" in (out + err).lower()
-    return True
-
-
-def _cuda_archs(acc: dict) -> list:
-    archs = []
-    for g in acc["gpus"]:
-        m = re.fullmatch(r"(\d+)\.(\d+)", (g.get("cc") or "").strip())
-        if m and (m.group(1) + m.group(2)) not in archs:
-            archs.append(m.group(1) + m.group(2))
-    return archs or ["native"]
-
-
-def _find_nvcc():
-    found = shutil.which("nvcc")
-    if found:
-        return found
-    for cand in sorted(glob.glob("/usr/local/cuda*/bin/nvcc"), reverse=True):
-        return cand
-    return None
+# --- getting a ds4-server binary ---------------------------------------------------
+def _ds4_binary_ok(binary: str) -> bool:
+    rc, out, err = _run([binary, "--help"], 90, None)
+    return "--ctx" in (out + err)
 
 
-def build_llama_cuda(archs: list):
-    """Build llama-server from source with CUDA (same recipe as the Unsloth guide + a single target arch).
-    Returns the path of the built binary, or None.
-
-    -DGGML_CUDA_NO_VMM=ON is essential on Kaggle: without it ggml-cuda links CUDA::cuda_driver (libcuda.so), and the
-    Kaggle image has the toolkit but NOT the libcuda.so stub, so CMake never creates that target and configure fails
-    with "Target ggml-cuda links to CUDA::cuda_driver but the target was not found" (ggml-cuda/CMakeLists.txt
-    lines 168-172; seen in a real T4x2 log). NO_VMM only drops the CUDA virtual-memory memory pool. A second attempt
-    also switches NCCL off, in case NCCL is what breaks the build/link."""
-    nvcc = _find_nvcc()
-    if not nvcc:
-        print("[llama][build] nvcc (CUDA toolkit) not found -- cannot build a CUDA llama-server here.", flush=True)
-        return None
-    cmake = shutil.which("cmake")
-    if not cmake:
-        _pip("install", "-q", "cmake", "ninja")
-        cmake = shutil.which("cmake")
-    if not cmake or not shutil.which("git"):
-        print("[llama][build] cmake/git missing -- cannot build.", flush=True)
-        return None
-    src, build = os.path.join(LLAMA_BUILD_ROOT, "llama.cpp"), os.path.join(LLAMA_BUILD_ROOT, "llama.cpp", "build")
-    log = os.path.join(LLAMA_BUILD_ROOT, "build.log")
-    env = os.environ.copy()
-    env["CUDACXX"] = nvcc
-    env["PATH"] = os.path.dirname(nvcc) + os.pathsep + env.get("PATH", "")
+def ensure_ds4_server():
+    """-> (binary, workdir). Order: GIJL_DS4_SERVER override, cached build, fresh `make <target>` of antirez/ds4."""
+    override = os.environ.get("GIJL_DS4_SERVER")
+    if override:
+        return override, os.path.dirname(override) or None
+    src = os.path.join(DS4_BUILD_ROOT, "ds4")
+    binary = os.path.join(src, "ds4-server")
+    if os.path.isfile(binary) and os.environ.get("GIJL_REBUILD_DS4") != "1" and _ds4_binary_ok(binary):
+        print(f"[ds4] using cached build {binary}", flush=True)
+        return binary, src
+    for tool in ("git", "make", "cc"):
+        if not shutil.which(tool):
+            raise RuntimeError(f"cannot build ds4: '{tool}' is not installed in this container "
+                               f"(or point GIJL_DS4_SERVER at a ds4-server binary you built elsewhere)")
+    log = os.path.join(DS4_BUILD_ROOT, "build.log")
+    os.makedirs(DS4_BUILD_ROOT, exist_ok=True)
+    shutil.rmtree(src, ignore_errors=True)            # a half-finished clone/build is worse than a fresh one
     t0 = time.time()
-    if not os.path.isdir(os.path.join(src, ".git")):
-        os.makedirs(LLAMA_BUILD_ROOT, exist_ok=True)
-        if _run_logged(["git", "clone", "--depth", "1", "https://github.com/ggml-org/llama.cpp", src], env, log, "llama-git", 600) != 0:
-            print(f"[llama][build] git clone failed:\n{_tail_file(log, 15)}", flush=True)
-            return None
-    toolkit_root = os.path.dirname(os.path.dirname(nvcc))
-    base = [cmake, "-S", src, "-B", build, "-DGGML_CUDA=ON", f"-DCMAKE_CUDA_ARCHITECTURES={';'.join(archs)}",
-            "-DCMAKE_BUILD_TYPE=Release", "-DBUILD_SHARED_LIBS=OFF", "-DLLAMA_CURL=OFF", "-DLLAMA_BUILD_TESTS=OFF",
-            "-DGGML_CUDA_NO_VMM=ON"]
-    if os.path.isfile(os.path.join(toolkit_root, "include", "cuda_runtime.h")):
-        base.append(f"-DCUDAToolkit_ROOT={toolkit_root}")
+    if _run_logged(["git", "clone", "--depth", "1"] + (["--branch", DS4_REF] if DS4_REF else []) + [DS4_REPO_URL, src],
+                   os.environ.copy(), log, "ds4-git", 600) != 0:
+        raise RuntimeError(f"git clone of {DS4_REPO_URL} failed:\n{_tail_file(log, 15)}")
     jobs = str(max(2, os.cpu_count() or 4))
-    for label, extra in (("CUDA, NO_VMM", []), ("CUDA, NO_VMM, NCCL off", ["-DGGML_CUDA_NCCL=OFF"])):
-        shutil.rmtree(build, ignore_errors=True)               # a failed configure leaves a poisoned CMake cache
-        print(f"[llama][build] attempt: {label} (archs {archs}); ~10-20 min once, then cached.", flush=True)
-        if _run_logged(base + extra, env, log, "llama-cmake", 900) != 0:
-            print(f"[llama][build] cmake configure failed ({label}):\n{_tail_file(log, 25)}", flush=True)
-            continue
-        if _run_logged([cmake, "--build", build, "--config", "Release", "-j", jobs, "--target", "llama-server"],
-                       env, log, "llama-build", LLAMA_BUILD_TIMEOUT_S) != 0:
-            print(f"[llama][build] build failed ({label}):\n{_tail_file(log, 25)}", flush=True)
-            continue
-        binary = os.path.join(build, "bin", "llama-server")
-        if os.path.isfile(binary):
-            print(f"[llama][build] done in {time.time() - t0:.0f}s ({label})", flush=True)
-            return binary
-    return None
-
-
-def download_llama_cpu_release():
-    """Prebuilt CPU llama-server from GitHub releases. The 'latest' release only carries nightly-tag.txt, which
-    names the tag holding the real binaries (llama-<tag>-bin-ubuntu-x64.tar.gz)."""
-    import requests
-    arch = "arm64" if platform.machine().lower() in ("aarch64", "arm64") else "x64"
-    base = "https://github.com/ggml-org/llama.cpp/releases"
-    hdr = {"User-Agent": _CRAWLER_UA}
-    pinned_tag = os.environ.get("GIJL_LLAMA_CPU_TAG", "").strip()
-    if pinned_tag:
-        # v8: bisecting knob, opt-in only (default behavior below is unchanged) -- added because DeepSeek-V4-Flash
-        # crashed llama-server on CPU with GGML_ASSERT(*cur_backend_id != -1) in ggml_backend_sched_split_graph
-        # during context/graph setup, a signature consistent with "this architecture's CPU backend coverage in
-        # llama.cpp is still incomplete" rather than anything wrong in this file -- DeepSeek-V4 support was still
-        # WIP/fork-based in llama.cpp as of its own GitHub support thread. "latest" always means whatever today's
-        # CI build is, so there is no way to test a different one without this. EXCLUSIVE on purpose (no silent
-        # fallback to latest): the point is to know for certain which build actually ran.
-        print(f"[llama] GIJL_LLAMA_CPU_TAG={pinned_tag} set -- using this exact release instead of latest", flush=True)
-        tags = [pinned_tag]
-    else:
-        tags = []
-        try:
-            latest = requests.get(f"{base}/latest", timeout=30, headers=hdr).url.rstrip("/").rsplit("/", 1)[-1]
-            nt = requests.get(f"{base}/download/{latest}/nightly-tag.txt", timeout=30, headers=hdr)
-            if nt.status_code == 200 and nt.text.strip():
-                tags.append(nt.text.strip().split()[0])
-            tags.append(latest)
-        except requests.RequestException as e:
-            raise RuntimeError(f"cannot reach GitHub releases: {_redact(str(e))[:200]}")
-    last_err = "no candidate tag"
-    for tag in tags:
-        dest = os.path.join(LLAMA_CPU_DIR, tag)
-        found = glob.glob(os.path.join(dest, "**", "llama-server"), recursive=True)
-        if found:
-            return found[0]
-        url = f"{base}/download/{tag}/llama-{tag}-bin-ubuntu-{arch}.tar.gz"
-        os.makedirs(dest, exist_ok=True)
-        archive = os.path.join(dest, "release.tar.gz")
-        try:
-            print(f"[llama] downloading CPU build {tag} ...", flush=True)
-            with requests.get(url, stream=True, timeout=(20, 120), headers=hdr) as r:
-                r.raise_for_status()
-                with open(archive, "wb") as f:
-                    for part in r.iter_content(chunk_size=1 << 20):
-                        f.write(part)
-            with tarfile.open(archive) as tf:
-                if hasattr(tarfile, "data_filter"):
-                    tf.extractall(dest, filter="data")
-                else:
-                    tf.extractall(dest)
-            os.remove(archive)
-        except Exception as e:  # noqa: BLE001
-            last_err = f"{tag}: {type(e).__name__}: {_redact(str(e))[:160]}"
-            continue
-        found = glob.glob(os.path.join(dest, "**", "llama-server"), recursive=True)
-        if found:
-            os.chmod(found[0], 0o755)
-            return found[0]
-    raise RuntimeError(f"could not obtain a prebuilt llama-server ({last_err})")
-
-
-def ensure_llama_server(acc: dict):
-    """-> (binary, lib_dir, kind). Order: GIJL_LLAMA_SERVER override, cached CUDA build, fresh CUDA build,
-    prebuilt CPU release (with a loud warning when a GPU exists but a CUDA build is impossible)."""
-    override = os.environ.get("GIJL_LLAMA_SERVER")
-    if override:
-        return override, os.path.dirname(override) or None, ("gpu" if acc["kind"] == "gpu" else "cpu")
-    if acc["kind"] == "gpu":
-        archs = _cuda_archs(acc)
-        cdir = os.path.join(LLAMA_BIN_DIR, "cuda-sm" + "-".join(archs))
-        cbin = os.path.join(cdir, "llama-server")
-        if os.path.isfile(cbin) and os.environ.get("GIJL_REBUILD_LLAMA") != "1":
-            for name in os.listdir(cdir):
-                try:
-                    os.chmod(os.path.join(cdir, name), 0o755)   # the hub does not preserve the executable bit
-                except OSError:
-                    pass
-            if _binary_ok(cbin, cdir, True):
-                print(f"[llama] using cached CUDA build {cbin}", flush=True)
-                return cbin, cdir, "gpu"
-            print("[llama] cached CUDA build is unusable here (different image/driver?) -- rebuilding", flush=True)
-        built = build_llama_cuda(archs)
-        if built:
-            shutil.rmtree(cdir, ignore_errors=True)
-            os.makedirs(cdir, exist_ok=True)
-            for name in os.listdir(os.path.dirname(built)):
-                fp = os.path.join(os.path.dirname(built), name)
-                if os.path.isfile(fp):
-                    shutil.copy2(fp, os.path.join(cdir, name))
-            if _binary_ok(cbin, cdir, True):
-                return cbin, cdir, "gpu"
-            print("[llama][build] the fresh build does not see a CUDA device", flush=True)
-        names = ", ".join(g["name"] for g in acc["gpus"])
-        if os.environ.get("GIJL_ALLOW_CPU_FALLBACK") != "1":
-            # This used to fall back silently: a T4x2 machine then ran a 30B model on 4 CPU cores (~1 token/s,
-            # 3 tok/s prefill) and nothing looked wrong for the first 15 minutes.
-            raise RuntimeError(
-                f"A GPU is present ({names}) but no CUDA llama-server could be built or loaded, so this would run on the CPU "
-                f"(~1 token/s). See the [llama][build] errors above and {os.path.join(LLAMA_BUILD_ROOT, 'build.log')}. "
-                f"Options: GIJL_LLAMA_SERVER=/path/to/your/cuda/llama-server, or GIJL_ALLOW_CPU_FALLBACK=1 to run on CPU anyway.")
-        print("[llama][WARNING] GIJL_ALLOW_CPU_FALLBACK=1 -> using the CPU build although a GPU is present (MUCH slower).", flush=True)
-    binary = download_llama_cpu_release()
-    if not _binary_ok(binary, os.path.dirname(binary), False):
-        raise RuntimeError(f"the downloaded llama-server does not run here: {binary}")
-    return binary, os.path.dirname(binary), "cpu"
-
-
-def load_tokenizer():
-    from transformers import AutoTokenizer
-    return AutoTokenizer.from_pretrained(MODEL_ID, token=os.environ.get("HF_TOKEN"))
+    print(f"[ds4][build] make {DS4_BUILD_TARGET} -j{jobs} (a few minutes, plain C) ...", flush=True)
+    if _run_logged(["make", "-C", src, DS4_BUILD_TARGET, "-j", jobs], os.environ.copy(), log, "ds4-build",
+                   DS4_BUILD_TIMEOUT_S) != 0 or not os.path.isfile(binary):
+        raise RuntimeError(f"ds4 build failed (make {DS4_BUILD_TARGET}):\n{_tail_file(log, 30)}")
+    if not _ds4_binary_ok(binary):
+        raise RuntimeError(f"the freshly built ds4-server does not run here: {binary}")
+    print(f"[ds4][build] done in {time.time() - t0:.0f}s: {binary}", flush=True)
+    return binary, src
 
 
 def _free_port() -> int:
@@ -1680,14 +1494,14 @@ def _free_port() -> int:
         return s.getsockname()[1]
 
 
-class LlamaServer:
-    """Owns the llama-server subprocess: starts it, waits for /health, restarts it if it dies, stops it at exit."""
+class Ds4Server:
+    """Owns the ds4-server subprocess: starts it, waits for /v1/models, restarts it if it dies, stops it at exit."""
 
-    def __init__(self, binary, lib_dir, model_path, kind, ctx, acc=None, model_gb=0.0, extra_args=None):
-        self.binary, self.lib_dir, self.model_path, self.kind, self.ctx = binary, lib_dir, model_path, kind, ctx
-        self.acc, self.model_gb, self.extra_args = acc or {}, model_gb, list(extra_args or [])
+    def __init__(self, binary, workdir, model_path, kind, ctx, model_gb=0.0, extra_args=None):
+        self.binary, self.workdir, self.model_path, self.kind, self.ctx = binary, workdir, model_path, kind, ctx
+        self.model_gb, self.extra_args = model_gb, list(extra_args or [])
         self.port, self.proc, self.restarts, self._help_text = _free_port(), None, 0, None
-        self.log_path = os.path.join(WORK_DIR, "llama-server.log")
+        self.log_path = os.path.join(WORK_DIR, "ds4-server.log")
         atexit.register(self.stop)
 
     @property
@@ -1696,30 +1510,14 @@ class LlamaServer:
 
     def _has(self, flag: str) -> bool:
         if self._help_text is None:
-            rc, out, err = _run([self.binary, "--help"], 90, _llama_env(self.lib_dir))
+            rc, out, err = _run([self.binary, "--help"], 90, None)
             self._help_text = out + err
         return flag in self._help_text
 
     def command(self) -> list:
-        cmd = [self.binary, "-m", self.model_path, "--host", "127.0.0.1", "--port", str(self.port), "-c", str(self.ctx)]
-        if self._has("--parallel"):
-            cmd += ["--parallel", "1"]                 # one stream: the whole context belongs to one slot
-        if self._has("--special"):
-            cmd += ["--special"]                       # emit special tokens so <atem:...> tool markup survives
-        if self._has("--no-context-shift"):
-            cmd += ["--no-context-shift"]              # error out instead of silently dropping the prompt's start
-        if self._has("--no-jinja"):
-            cmd += ["--no-jinja"]                      # we render the template ourselves: /completion stays raw pass-through
-        if self._has("--reasoning-format"):
-            cmd += ["--reasoning-format", "none"]      # never move 'thinking' out of the text we parse
-        threads = "16"
-        cmd += ["--threads", threads]
-        if self.kind == "gpu":
-            if self.model_gb and self.model_gb + 2.5 <= self.acc.get("vram_gb", 0) * 0.92:
-                cmd += ["-ngl", "999"]                 # fits entirely: offload every layer
-            # else: leave -ngl unset so llama.cpp's default --fit adjusts layers to the free VRAM
-        else:
-            cmd += ["-ngl", "0"]
+        cmd = [self.binary, "-m", self.model_path, "--host", "127.0.0.1", "--port", str(self.port), "--ctx", str(self.ctx)]
+        if self._has("--threads"):
+            cmd += ["--threads", "16"]
         return cmd + self.extra_args
 
     def tail(self, n: int = 25) -> str:
@@ -1729,40 +1527,34 @@ class LlamaServer:
         import requests
         os.makedirs(os.path.dirname(self.log_path), exist_ok=True)
         cmd = self.command()
-        print("[llama] starting: " + " ".join(cmd), flush=True)
+        print("[ds4] starting: " + " ".join(cmd), flush=True)
         self._log = open(self.log_path, "ab")
-        self.proc = subprocess.Popen(cmd, stdout=self._log, stderr=subprocess.STDOUT, env=_llama_env(self.lib_dir),
+        self.proc = subprocess.Popen(cmd, stdout=self._log, stderr=subprocess.STDOUT, cwd=self.workdir,
                                      start_new_session=True)
         t0, last = time.time(), time.time()
-        while time.time() - t0 < LLAMA_LOAD_TIMEOUT_S:
+        while time.time() - t0 < DS4_LOAD_TIMEOUT_S:
             if self.proc.poll() is not None:
-                raise RuntimeError(f"llama-server exited during startup (code {self.proc.returncode}). Log tail:\n{self.tail(40)}")
+                raise RuntimeError(f"ds4-server exited during startup (code {self.proc.returncode}). Log tail:\n{self.tail(40)}")
             try:
-                r = requests.get(self.base_url + "/health", timeout=5)
-                if r.status_code == 200 and r.json().get("status") == "ok":
-                    print(f"[llama] server ready after {time.time() - t0:.0f}s on port {self.port}", flush=True)
-                    self._print_offload_summary()
+                r = requests.get(self.base_url + "/v1/models", timeout=5)
+                if r.status_code == 200:
+                    print(f"[ds4] server ready after {time.time() - t0:.0f}s on port {self.port}", flush=True)
                     return
             except (requests.RequestException, ValueError):
                 pass
             if time.time() - last > 30:
-                print(f"[llama] loading model ... {time.time() - t0:.0f}s", flush=True)
+                print(f"[ds4] loading model ... {time.time() - t0:.0f}s", flush=True)
                 last = time.time()
             time.sleep(2)
-        raise TimeoutError(f"llama-server not ready after {LLAMA_LOAD_TIMEOUT_S}s. Log tail:\n{self.tail(40)}")
-
-    def _print_offload_summary(self):
-        lines = [l for l in self.tail(400).splitlines() if re.search(r"offload|CUDA|fit|n_ctx|KV", l)]
-        for l in lines[-6:]:
-            print("[llama] " + l[-200:], flush=True)
+        raise TimeoutError(f"ds4-server not ready after {DS4_LOAD_TIMEOUT_S}s. Log tail:\n{self.tail(40)}")
 
     def ensure_running(self):
         if self.proc is not None and self.proc.poll() is None:
             return
         self.restarts += 1
-        print(f"[llama][WARNING] server is not running (restart #{self.restarts}). Log tail:\n{self.tail(15)}", flush=True)
+        print(f"[ds4][WARNING] server is not running (restart #{self.restarts}). Log tail:\n{self.tail(15)}", flush=True)
         if self.restarts > 3:
-            raise RuntimeError("llama-server keeps dying; giving up")
+            raise RuntimeError("ds4-server keeps dying; giving up")
         self.port = _free_port()
         self.start()
 
@@ -1780,78 +1572,75 @@ class LlamaServer:
                 pass
 
 
-class LlamaCppClient:
-    """Same interface as ModelClient.chat(): renders the model's OWN chat template (with tools) via the HF
-    tokenizer and streams /completion from llama-server.
-    - BOS: llama-server adds BOS itself, so a BOS already in the rendered text would double it (verified);
-      the leading BOS string is stripped when the server adds one.
-    - closing the HTTP connection cancels generation server-side (verified), so a timeout really frees the
-      slot -- unlike the old transformers thread that kept running.
-    - prompt caching is on: each scouting turn only prefills what is new."""
-
-    def __init__(self, server: LlamaServer, tokenizer, kind: str):
-        self.server, self.tok, self.kind, self.hardware = server, tokenizer, kind, kind
-        self.extra_payload = {}            # (tests / advanced use) merged into every /completion request
-        self.ctx = self._read_ctx()
-        self.server_adds_bos = self._probe_add_bos()
-
-    def _read_ctx(self) -> int:
-        import requests
-        try:
-            n = requests.get(self.server.base_url + "/props", timeout=20).json()["default_generation_settings"]["n_ctx"]
-            return int(n)
-        except Exception:  # noqa: BLE001
-            return self.server.ctx
-
-    def _tokenize(self, text: str, add_special: bool) -> list:
-        import requests
-        r = requests.post(self.server.base_url + "/tokenize", timeout=120,
-                          json={"content": text, "add_special": add_special, "parse_special": True})
-        r.raise_for_status()
-        return r.json()["tokens"]
-
-    def _probe_add_bos(self) -> bool:
-        try:
-            return len(self._tokenize("a", True)) == len(self._tokenize("a", False)) + 1
-        except Exception:  # noqa: BLE001
-            return False
-
-    def _prepare_prompt(self, prompt: str) -> str:
-        bos = getattr(self.tok, "bos_token", None)
-        if self.server_adds_bos and bos and prompt.startswith(bos):
-            prompt = prompt[len(bos):]
-        return prompt
+class Ds4Client:
+    """Same interface as ModelClient.chat(): streams POST /v1/chat/completions from ds4-server (OpenAI-style SSE).
+    - ds4 renders the DeepSeek chat template itself, so `messages` go over as plain chat. Messages with role='tool'
+      are folded into user messages (the same fallback _render_chat uses) and consecutive same-role messages are
+      merged, so the server only ever sees a clean system/user/assistant alternation.
+    - thinking follows _CHAT_TEMPLATE_EXTRA_KWARGS["enable_thinking"] (False for this worker) via the request's
+      `think` field, which is ds4's documented switch.
+    - a separately streamed reasoning_content is NOT part of the returned text (only visible `content` is).
+    - closing the HTTP connection is how a timeout cancels generation (assumed to be honored by ds4-server like any
+      SSE server -- NOT verified here; if the log shows generation continuing after a [gen][TIMEOUT], that is why)."""
+
+    def __init__(self, server: Ds4Server, kind: str):
+        self.server, self.kind, self.hardware = server, kind, kind
+        self.tok = None                    # no local tokenizer: ds4 tokenizes and renders the template itself
+        self.ctx = server.ctx
+        self.extra_payload = {}            # (tests / advanced use) merged into every request
+        self._dropped = set()              # optional request fields this server build rejected with HTTP 400
+
+    @staticmethod
+    def _wire_messages(messages) -> list:
+        out = []
+        for m in messages:
+            role, content = m.get("role", "user"), str(m.get("content", "") or "")
+            if role == "tool":
+                role, content = "user", f"[tool result from {m.get('name', 'tool')}]\n{content}"
+            if out and out[-1]["role"] == role and role != "system":
+                out[-1]["content"] += "\n\n" + content
+            else:
+                out.append({"role": role, "content": content})
+        return out
 
     def chat(self, messages, max_new_tokens=GEN_MAX_NEW_TOKENS, temperature=0.4, tools=None):
         import requests
-        tool_mode = bool(tools)
         self.server.ensure_running()
-        prompt = self._prepare_prompt(_render_chat(self.tok, messages, tools))
-        try:
-            n_prompt = len(self._tokenize(prompt, True))
-        except requests.RequestException:
-            n_prompt = -1
-        if n_prompt >= 0 and n_prompt + 256 > self.ctx:
-            print(f"[llama] prompt is {n_prompt} tokens with ctx={self.ctx} -- refusing to generate", flush=True)
-            return {"content": "", "error": "context_too_long", "timed_out": False, "prompt_tokens": n_prompt}
-        n_predict = max_new_tokens if n_prompt < 0 else max(64, min(max_new_tokens, self.ctx - n_prompt - 32))
-        payload = {"prompt": prompt, "n_predict": n_predict, "temperature": temperature, "top_p": TOP_P, "top_k": TOP_K,
-                   "stream": True, "cache_prompt": True}
+        payload = {"model": DS4_MODEL_ALIAS, "messages": self._wire_messages(messages), "stream": True,
+                   "temperature": temperature, "top_p": TOP_P, "top_k": TOP_K, "max_tokens": max_new_tokens,
+                   "think": bool(_CHAT_TEMPLATE_EXTRA_KWARGS.get("enable_thinking", True)),
+                   "stream_options": {"include_usage": True}}
         payload.update(self.extra_payload)
+        for k in self._dropped:
+            payload.pop(k, None)
 
         gap = GEN_PER_CHUNK_GAP_TIMEOUT_S.get(self.kind, 300)
         cap = GEN_MAX_TOTAL_S.get(self.kind, 900)
-        chunks, final, err, timed_out, resp = [], None, None, False, None
+        chunks, usage, err, timed_out, resp, finish = [], {}, None, False, None, None
         start = last_data = last_hb = time.time()
         try:
-            resp = requests.post(self.server.base_url + "/completion", json=payload, stream=True, timeout=(15, gap))
-            if resp.status_code != 200:
+            for _attempt in range(2):
+                resp = requests.post(self.server.base_url + "/v1/chat/completions", json=payload, stream=True,
+                                     timeout=(15, gap))
+                if resp.status_code == 200:
+                    break
                 body = resp.text[:400]
-                err = "context_too_long" if "exceed_context_size" in body else f"HTTP {resp.status_code}: {_redact(body)}"
-            else:
-                # bytes -> utf-8 ourselves: text/event-stream without a charset would be decoded as latin-1 (Arabic!).
-                # chunk_size=1: the default (512) blocks until 512 bytes arrive, so with un-chunked framing (HTTP/1.0,
-                # some proxies) keep-alive pings could hide a stalled server for minutes (found by the fake-server test).
+                resp.close()
+                optional = [k for k in ("top_k", "stream_options") if k in payload and k in body]
+                if resp.status_code == 400 and optional and _attempt == 0:
+                    for k in optional:             # this build does not know the field -> drop it for good and retry
+                        self._dropped.add(k)
+                        payload.pop(k, None)
+                    print(f"[ds4] server rejected {optional}; retrying without", flush=True)
+                    continue
+                low = body.lower()
+                err = ("context_too_long" if resp.status_code == 400 and
+                       ("context" in low or "too long" in low or "exceed" in low) else
+                       f"HTTP {resp.status_code}: {_redact(body)}")
+                resp = None
+                break
+            if resp is not None:
+                # bytes -> utf-8 ourselves (Arabic!); chunk_size=1 so keep-alive pings can't hide a stalled server.
                 for raw in resp.iter_lines(chunk_size=1):
                     now = time.time()
                     if now - start > cap:
@@ -1864,9 +1653,12 @@ class LlamaCppClient:
                         break
                     line = raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw
                     if not line.startswith("data:"):
-                        continue                       # SSE keep-alive pings / blank lines
+                        continue                       # SSE keep-alive pings / blank lines / comments
+                    data = line[5:].strip()
+                    if data == "[DONE]":
+                        break
                     try:
-                        d = json.loads(line[5:].strip())
+                        d = json.loads(data)
                     except ValueError:
                         continue
                     if "error" in d:
@@ -1874,19 +1666,20 @@ class LlamaCppClient:
                         err = _redact(str(e.get("message", e)) if isinstance(e, dict) else str(e))
                         break
                     last_data = now
-                    if d.get("content"):
-                        chunks.append(d["content"])
-                    if d.get("stop"):
-                        final = d
-                        break
+                    if isinstance(d.get("usage"), dict):
+                        usage = d["usage"]
+                    for ch in d.get("choices") or []:
+                        delta = ch.get("delta") or {}
+                        if delta.get("content"):
+                            chunks.append(delta["content"])
+                        finish = ch.get("finish_reason") or finish
                     if now - last_hb > 30:
                         print(f"[gen] {sum(len(c) for c in chunks)} chars, {now - start:.0f}s elapsed ({self.kind})", flush=True)
                         last_hb = now
         except requests.RequestException as e:
-            # While streaming, requests surfaces a socket read timeout as ConnectionError("... Read timed out"),
-            # NOT as ReadTimeout -- so match on the message too.
+            # While streaming, requests surfaces a socket read timeout as ConnectionError("... Read timed out").
             if isinstance(e, requests.exceptions.Timeout) or "timed out" in str(e).lower():
-                print(f"[gen][TIMEOUT] llama-server sent nothing for {gap}s -- treating as stuck.", flush=True)
+                print(f"[gen][TIMEOUT] ds4-server sent nothing for {gap}s -- treating as stuck.", flush=True)
                 timed_out = True
             else:
                 err = f"{type(e).__name__}: {_redact(str(e))[:200]}"
@@ -1894,52 +1687,43 @@ class LlamaCppClient:
             if resp is not None:
                 resp.close()                          # closing the connection cancels generation on the server
 
-        cached = gen = 0
-        if final:
-            t = final.get("timings") or {}
-            cached, gen = int(t.get("cache_n") or 0), int(final.get("tokens_predicted") or 0)
-            print(f"[llama] prompt {final.get('tokens_evaluated')} tok (cached {cached}) prefill "
-                  f"{t.get('prompt_per_second') or 0:.0f} tok/s | generated {gen} tok @ "
-                  f"{t.get('predicted_per_second') or 0:.1f} tok/s | stop={final.get('stop_type')}", flush=True)
+        n_prompt, gen = int(usage.get("prompt_tokens") or -1), int(usage.get("completion_tokens") or 0)
+        el = max(time.time() - start, 1e-6)
+        print(f"[ds4] prompt {n_prompt if n_prompt >= 0 else '?'} tok | generated "
+              f"{gen if gen else '?'} tok in {el:.0f}s{f' (~{gen / el:.1f} tok/s incl. prefill)' if gen else ''} "
+              f"| finish={finish}", flush=True)
         if err and err != "context_too_long":
-            print(f"[llama] request failed: {err}", flush=True)
-        text = _strip_special_tokens(self.tok, "".join(chunks), keep_atem=tool_mode)
-        return {"content": text, "error": err, "timed_out": timed_out, "prompt_tokens": n_prompt,
-                "cached_tokens": cached, "gen_tokens": gen}
+            print(f"[ds4] request failed: {err}", flush=True)
+        return {"content": "".join(chunks), "error": err, "timed_out": timed_out, "prompt_tokens": n_prompt,
+                "cached_tokens": 0, "gen_tokens": gen}
 
     def warmup(self):
         """One tiny generation at startup: proves the whole path works and prints the real speed."""
         r = self.chat([{"role": "user", "content": "Reply with the single word: ready"}], max_new_tokens=24, temperature=0.0)
         ok = bool(r["content"].strip()) and not r["error"]
-        print(f"[llama] warmup {'ok' if ok else 'FAILED'}: {r['content'].strip()[:60]!r} {r['error'] or ''}", flush=True)
+        print(f"[ds4] warmup {'ok' if ok else 'FAILED'}: {r['content'].strip()[:60]!r} {r['error'] or ''}", flush=True)
         return ok
 
 
-def start_llama_backend():
-    """-> (client, server). Model download, binary build/download and tokenizer load run in parallel."""
+def start_ds4_backend():
+    """-> (client, server). Model download and ds4 build run in parallel."""
     from concurrent.futures import ThreadPoolExecutor
     acc = detect_accelerators()
-    gpu_desc = ", ".join(f"{g['name']} {g['mem_gb']:.1f}GB (sm{g['cc'] or '?'})" for g in acc["gpus"]) or "none"
-    print(f"[hardware] llama.cpp backend | GPUs: {gpu_desc} | RAM {acc['ram_gb']:.1f}GB | CPUs {acc['cpus']}", flush=True)
-    if acc["kind"] == "cpu":
-        print("[hardware][WARNING] no NVIDIA GPU found -> CPU inference, roughly 1 token/s for a 30B model. "
-              "On Kaggle choose Accelerator = GPU T4 x2.", flush=True)
-        if acc["tpu_hint"]:
-            print("[hardware][WARNING] a TPU VM was detected: llama.cpp cannot use TPUs, and Kaggle stops TPU sessions that do "
-                  "not use the TPU (your last TPU log ended with Exit 137 after 2h idle).", flush=True)
-    with ThreadPoolExecutor(max_workers=3) as ex:
-        f_model, f_bin, f_tok = ex.submit(download_gguf, acc), ex.submit(ensure_llama_server, acc), ex.submit(load_tokenizer)
-        (model_path, meta), (binary, lib_dir, kind), tok = f_model.result(), f_bin.result(), f_tok.result()
-    server = LlamaServer(binary, lib_dir, model_path, kind, LLAMA_CTX, acc, meta["size_gb"])
-    print(f"[llama] inference device: {'GPU (CUDA)' if kind == 'gpu' else 'CPU'}", flush=True)
+    kind = "cpu" if DS4_BUILD_TARGET == "cpu" else "gpu"
+    print(f"[hardware] ds4 backend (make {DS4_BUILD_TARGET}) | RAM {acc['ram_gb']:.1f}GB | CPUs {acc['cpus']}", flush=True)
+    if kind == "cpu" and acc["kind"] == "gpu":
+        print("[hardware][WARNING] an NVIDIA GPU is present but GIJL_DS4_BUILD=cpu (default) -> CPU inference. "
+              "For GPU set GIJL_DS4_BUILD=cuda-generic (builds ds4's CUDA path).", flush=True)
+    with ThreadPoolExecutor(max_workers=2) as ex:
+        f_model, f_bin = ex.submit(download_gguf, acc), ex.submit(ensure_ds4_server)
+        (model_path, meta), (binary, workdir) = f_model.result(), f_bin.result()
+    server = Ds4Server(binary, workdir, model_path, kind, DS4_CTX, meta["size_gb"])
+    print(f"[ds4] inference device: {'GPU' if kind == 'gpu' else 'CPU'}", flush=True)
     server.start()
-    if kind == "gpu" and not re.search(r"offloaded|CUDA\d", server.tail(600)):
-        print("[llama][WARNING] the server log does not mention CUDA/offloading -- it may be running on CPU. "
-              f"Check {server.log_path}", flush=True)
-    client = LlamaCppClient(server, tok, kind)
-    print(f"[llama] context per request: {client.ctx} tokens | server adds BOS: {client.server_adds_bos}", flush=True)
+    client = Ds4Client(server, kind)
+    print(f"[ds4] context per request: {client.ctx} tokens", flush=True)
     if not client.warmup():
-        print("[llama][WARNING] warmup produced no text; continuing, but check llama-server.log", flush=True)
+        print("[ds4][WARNING] warmup produced no text; continuing, but check ds4-server.log", flush=True)
     return client, server
 
 
@@ -4384,7 +4168,7 @@ def _sync_patterns() -> list:
     # filenames (MANIFEST.json/README.md) needed a MODEL_TAG suffix -- done where they are defined above.
     pats = ([f"{m['folder']}/*" for m in DATA_TYPES.values()]
             + [f"state/{MODEL_TAG}/*", os.path.basename(MANIFEST_PATH), os.path.basename(README_PATH)])
-    return pats + (["bin/*"] if CACHE_LLAMA_BINARY and BACKEND == "llamacpp" else [])
+    return pats
 
 
 def _with_retry(fn, what: str, tries: int = 3, base_delay: float = 2.0):
@@ -4418,8 +4202,7 @@ def restore_from_hf():
         # other two workers' data down into every container unnecessarily.
         _with_retry(lambda: snapshot_download(
             repo_id=TARGET_DATASET_REPO, repo_type="dataset", token=token, local_dir=WORK_DIR,
-            allow_patterns=[f"{m['folder']}/{MODEL_TAG}/*" for m in DATA_TYPES.values()] + [f"state/{MODEL_TAG}/*"]
-            + (["bin/*"] if CACHE_LLAMA_BINARY and BACKEND == "llamacpp" else [])), "restore")
+            allow_patterns=[f"{m['folder']}/{MODEL_TAG}/*" for m in DATA_TYPES.values()] + [f"state/{MODEL_TAG}/*"]), "restore")
         print(f"[restore] pulled prior state from {TARGET_DATASET_REPO} (worker={MODEL_TAG})", flush=True)
     except Exception as e:
         print(f"[restore] nothing to restore yet ({_redact(str(e))})", flush=True)
@@ -4448,16 +4231,16 @@ def sync_to_hf():
 # 14. Main loop -- single process, runs until the container stops (or GIJL_SESSION_BUDGET_S, if set)
 # ---------------------------------------------------------------------------
 def _start_backend():
-    """-> (client, close). close() stops llama-server (no-op for the transformers backend)."""
+    """-> (client, close). close() stops ds4-server (no-op for the transformers backend)."""
     if BACKEND == "hf":
         hardware = detect_hardware()
         print(f"[hardware] transformers backend, selected: {hardware}", flush=True)
         if hardware == "tpu":
             print("[hardware][WARNING] the transformers path on TPU was ~3 chars/s and hung in XLA compilation (see your TPU "
-                  "log). Use the default llama.cpp backend with a GPU accelerator instead.", flush=True)
+                  "log). Use the default ds4 backend instead.", flush=True)
         model, tok = load_model(hardware)
         return ModelClient(model, tok, hardware), (lambda: None)
-    client, server = start_llama_backend()
+    client, server = start_ds4_backend()
     return client, server.stop
 
 
@@ -4537,7 +4320,7 @@ def main():
             print(f"[cycle] took {time.time() - t_cycle:.0f}s. " + _status_line(cycle, state), flush=True)
             maybe_sync(state)
         if smoke:
-            print("[smoke] one cycle finished -- see the [llama] tok/s lines and [gen]/[write] lines above. "
+            print("[smoke] one cycle finished -- see the [ds4] tok/s lines and [gen]/[write] lines above. "
                   "Unset GIJL_SMOKE for the full run.", flush=True)
     finally:
         try: