"""Run inference against a supervised ``llama-server`` over loopback HTTP. This replaces spawning ``llama-cli`` per request, and it is a better design rather than a way around a bug. **Why the server, not the CLI.** From llama.cpp b10451 the CLI is a chat front end: it prints a logo, build metadata and a slash-command list to stdout, then blocks on stdin. ``-st`` stops the hang but the furniture still lands in the response, and parsing it out means depending on a format that changes between builds. The server's banner goes to its own log and never enters the response body, so there is nothing to parse. Three things follow that the CLI could not give us at all: * **token counts**, which per-turn usage reporting needs; * **streaming**, which live agent turns and traces need; * **an OpenAI-compatible endpoint**, which is what DSPy speaks natively, so the RLM harness points at a URL instead of wrapping a subprocess. And the model stays resident across queued requests instead of reloading per request, which on a project measuring energy per run is not a footnote. **The costs, handled rather than waved past.** It is a long-lived child, so it is supervised and killed with the worker. It binds a port, so it binds ``127.0.0.1`` on an ephemeral port, consistent with :mod:`distinct_protocol.netpolicy`. It is spawned through :func:`distinct_agent.isolation.spawn_restricted`, so it sits inside the Job Object rather than beside it. """ from __future__ import annotations import json import re import socket import subprocess import threading import time import urllib.error import urllib.request from collections import deque from collections.abc import Callable, Mapping from pathlib import Path from typing import Any from distinct_protocol import PHASE_GENERATING, PHASE_LOADING_MODEL, STEP_PHASE from .energy import EnergyMeter, UnavailableEnergyMeter from .isolation import ProcessRestriction, spawn_restricted from .models import DiscoveredModel from .runners import ( InferenceResult, RunnerCancelled, RunnerError, RunnerTimedOut, RunnerUnavailable, build_llama_environment, ) LOOPBACK = "127.0.0.1" #: The share of *free* VRAM an offload is allowed to plan for. The rest is #: left to the display, to whatever else is on the card, and to llama.cpp's own #: scratch buffers, which are not in the arithmetic below. VRAM_BUDGET_SHARE = 0.72 #: Fixed cost of having a CUDA context and llama.cpp's compute buffers on the #: card at all, before a single layer is placed there. VRAM_FIXED_OVERHEAD = 320 * 1024 * 1024 def plan_offload( *, model_bytes: int, shape: Any, free_vram_bytes: int, context_tokens: int, ) -> tuple[int, ...]: """How many layers to put on the GPU, most first, never more than fits. ASKING FOR MORE THAN FITS IS NOT A FAILURE ON WINDOWS, AND THAT IS THE PROBLEM. The obvious design is to ask for everything and step down when the allocation fails. It is what this did first, and on the benchmark machine -- a 4 GB laptop card with 2.3 GB free and a 4.5 GB model -- it made every single workload time out where they had previously completed. Nothing failed. The WDDM driver backed the excess with system memory, the allocation succeeded, the server reported itself healthy, and then every token dragged weights back across the PCIe bus. The result was several times slower than the CPU-only build it replaced, and it looked like a hang rather than a misconfiguration. So the ceiling is worked out before anything is asked for. Two costs per offloaded layer: its weights, and its slice of the key/value cache, which at a 4096-token context can exceed the weights and is the term that a weights-only estimate misses. Both come out of the model file's own header. A model whose header cannot be read plans nothing and runs on the CPU, which is where it ran before any of this existed. """ if not shape.known or model_bytes <= 0 or free_vram_bytes <= 0: return (0,) usable = int(free_vram_bytes * VRAM_BUDGET_SHARE) - VRAM_FIXED_OVERHEAD if usable <= 0: return (0,) # The +1 stands in for the embedding and output tensors, which are not a # layer but do take room when they are offloaded with one. weights_per_layer = model_bytes / (shape.layers + 1) per_layer = weights_per_layer + shape.kv_bytes_per_layer(context_tokens) if per_layer <= 0: return (0,) top = max(0, min(shape.layers, int(usable // per_layer))) if top <= 0: return (0,) rungs: list[int] = [] for value in (top, top * 3 // 4, top // 2, top // 4, 0): if value not in rungs: rungs.append(value) return tuple(rungs) #: What llama.cpp says when the accelerator has not got the room. Matched on #: the stderr tail, and only to decide whether offloading *less* is worth #: trying: a failure that is not about memory will fail identically on fewer #: layers, so the ladder stops rather than walking every rung to the same end. _SHORT_OF_MEMORY = re.compile( r"out of memory|failed to allocate|cudaMalloc|unable to allocate|" r"insufficient memory|ggml_backend_.*alloc.*failed|not enough (?:VRAM|memory)", re.IGNORECASE, ) def _short_of_memory(said: str) -> bool: return bool(said) and bool(_SHORT_OF_MEMORY.search(said)) #: What a per-run energy figure covers. #: #: Keeping the model resident across a queue makes this a real question rather #: than a pedantic one. Loading the weights costs energy once; the naive #: measurement charges the whole of it to whichever run happened to arrive #: first and nothing to the rest. Two runs of the identical prompt then differ #: by the load cost purely because of queue position, and the figure misleads #: in opposite directions depending on which one you look at. #: #: So load is never folded into a run. Every run reports #: ``inference-only``, and the load cost is measured once, attributed to the #: model's residency, and reported beside the run that triggered it. ENERGY_BOUNDARY_INFERENCE_ONLY = "inference-only" class LlamaServerUnavailable(RunnerUnavailable): """The server binary is missing, or it never became healthy.""" def _free_port() -> int: with socket.socket() as probe: probe.bind((LOOPBACK, 0)) return int(probe.getsockname()[1]) class _ProcessLog: """The last few kilobytes llama-server said, kept for its epitaph. A thread rather than a poll, because the pipe has to be drained: a child whose stderr buffer fills blocks on write, and a blocked inference server looks exactly like a hung one. """ #: Enough to hold a start-up banner or a stack of allocator complaints, #: little enough that a week-long residency cannot grow into it. LIMIT = 8192 def __init__(self) -> None: self._chunks: deque[str] = deque() self._size = 0 self._lock = threading.Lock() self._thread: threading.Thread | None = None def follow(self, stream: Any) -> None: if stream is None: return def pump() -> None: try: for raw in iter(stream.readline, b""): text = raw.decode("utf-8", "replace") with self._lock: self._chunks.append(text) self._size += len(text) while self._size > self.LIMIT and len(self._chunks) > 1: self._size -= len(self._chunks.popleft()) except (ValueError, OSError): # The pipe closed under us, which is what happens when the # process exits. Not an error: the exit is the news. pass self._thread = threading.Thread(target=pump, name="llama-stderr", daemon=True) self._thread.start() def tail(self, *, limit: int = 600) -> str: with self._lock: text = "".join(self._chunks) lines = [line.strip() for line in text.splitlines() if line.strip()] joined = " | ".join(lines[-6:]) return joined[-limit:] if joined else "" class LlamaServerRunner: """Supervise one ``llama-server`` and answer requests against it. One server per model. ``base_url`` is exposed so the DSPy RLM harness can point at the same endpoint rather than starting a second copy. """ name = "llama.cpp-server" #: The calibration generation. Short, deterministic and long enough for #: llama.cpp's own tokens-per-second figure to settle. PROBE_PROMPT = "Count from one to twenty, separated by commas." PROBE_TOKENS = 96 PROBE_TIMEOUT = 180.0 def __init__( self, executable: Path | str, *, context_length: int = 4096, startup_timeout_seconds: float = 240.0, recovery_seconds: float = 20.0, gpu_layers: tuple[int, ...] | None = None, restriction: ProcessRestriction | None = None, clock: Callable[[], float] = time.monotonic, energy_meter: EnergyMeter | None = None, ) -> None: self.executable = str(executable) self.context_length = int(context_length) self.startup_timeout_seconds = float(startup_timeout_seconds) #: How long a server that has stopped answering is given to come back #: before it is assumed dead and restarted. Long enough to cover a #: cancelled task being unwound, short enough that a genuinely dead #: server is not waited on. self.recovery_seconds = float(recovery_seconds) #: An explicit ladder, or ``None`` to work one out per model from the #: card's free memory and the model's own header. self.gpu_layers = tuple(gpu_layers) if gpu_layers is not None else None self.restriction = restriction or ProcessRestriction(mode="auto") self._clock = clock # Used only to price model loading, which is a property of residency # rather than of any one run. Runs are metered by the caller. self._energy_meter = energy_meter or UnavailableEnergyMeter("no meter supplied") self._lock = threading.Lock() self._process: subprocess.Popen | None = None self._guard: Any = None self._port: int | None = None self._model_path: Path | None = None self._model_id: str | None = None self.isolation: Any = None #: Energy spent loading the resident model, or ``None`` if unmeasured. #: Never zero for "unknown": absence and a measured zero are different. self.load_energy_joules: float | None = None self.load_seconds: float | None = None #: How many layers the resident server actually got onto the GPU. #: ``None`` until something is loaded; ``0`` is a real answer meaning #: everything is on the CPU, and is not the same as "not measured". self.gpu_layers_used: int | None = None #: Whether this build honours a JSON-schema constraint. ``None`` until #: one has been tried; ``False`` after a build refuses one, which stops #: every later request paying for the same refusal. self.supports_schema: bool | None = None #: Why it refused, when it did. Empty otherwise. self.schema_refusal: str = "" #: What calibration found, when it ran in this process. Empty when the #: answer came from the cache, which is the usual case. self.offload_measurements: tuple[Any, ...] = () #: Called with a sentence while calibration runs, so a worker that is #: busy for a minute says what it is doing. self._notify: Callable[[str], None] | None = None self._load_charged = False # -- lifecycle ---------------------------------------------------- @property def available(self) -> bool: return bool(self.executable and Path(self.executable).is_file()) @property def base_url(self) -> str: if self._port is None: raise LlamaServerUnavailable("server is not running") return f"http://{LOOPBACK}:{self._port}" def ensure_started(self, model: DiscoveredModel) -> None: """Start the server for ``model``, restarting if the model changed. Serialised: two jobs arriving together must not race two servers onto two ports and leave one orphaned. """ with self._lock: if self._process is not None and self._process.poll() is None: if self._model_path == model.path and self._answering(): return # ALIVE IS NOT THE SAME AS ANSWERING, AND THE DIFFERENCE COST A # RUN EVERY TIME. # # This checked only that the process existed. When llama-server # lost its listener but stayed up -- which is what it does on # some failures rather than exiting -- the check passed, the # request went to a closed port, and the run failed with # "connection actively refused". The run AFTER a crash was # therefore lost too: the first one to see the dead socket paid # for discovering it. Asking the health endpoint costs a # millisecond on loopback and turns that into a restart. self._stop_locked() self._start_locked(model) def _await_health(self, seconds: float) -> bool: """Poll ``/health`` until the server answers, or the time runs out. Distinct from :meth:`_wait_for_health`, which is start-up: this is for a server that was healthy a moment ago and may only be busy. """ deadline = self._clock() + max(0.0, seconds) while True: if self._process is not None and self._process.poll() is not None: return False if self._answering(): return True if self._clock() >= deadline: return False time.sleep(0.25) def _retry_after_transport_loss( self, model: DiscoveredModel, prompt: str, lost: _TransportLost, *, max_tokens: int, temperature: float, timeout: float, schema: Mapping[str, Any] | None = None, ) -> tuple[Mapping[str, Any], str]: """One request lost its connection. Get the server back and ask again. A DROPPED CONNECTION IS NOT A WRONG ANSWER, AND WAS BEING TREATED AS ONE. ``llama-server became unreachable: [WinError 10054]`` ended a run and threw away every tool result it had already earned. But the reset does not mean the model was wrong or the request was bad; it means the socket died, usually because the previous request was cancelled and the server was still releasing the slot, occasionally because the server crashed and has to be brought back. Neither is a reason to lose the work. So: wait for the server to answer ``/health`` again, and if it will not, restart it -- and either way ask once more. Once, not in a loop. A second loss with a healthy server in between is a real fault and deserves to be reported rather than papered over, and a retry loop around an expensive generation is how a worker burns an evening of somebody's electricity on a request that was never going to succeed. """ if not self._await_health(self.recovery_seconds): with self._lock: self._stop_locked() self._start_locked(model) try: return self._generate( prompt, max_tokens=max_tokens, temperature=temperature, timeout=timeout, schema=schema, ) except _TransportLost as again: raise RunnerError(f"{again} (it had already been recovered once)") from again def _answering(self) -> bool: """Whether the server on our port is actually serving.""" if self._port is None: return False try: with urllib.request.urlopen( f"http://{LOOPBACK}:{self._port}/health", timeout=2 ) as response: return response.status == 200 except (urllib.error.URLError, OSError): return False def evict(self, model_id: str | None = None) -> bool: """Shut the server down when its model is evicted from the cache. Without this, a cache that evicts a model leaves its server resident, the worker accumulates loaded models, and the energy figures the product exists to report become nonsense. Returns whether anything was stopped, so an eviction policy can tell a hit from a miss. """ with self._lock: if self._process is None: return False if model_id is not None and self._model_id != model_id: return False self._stop_locked() return True @property def resident_model_id(self) -> str | None: """Which model is currently loaded, if any.""" return self._model_id def _start_locked(self, model: DiscoveredModel) -> None: """Bring the server up, on the GPU if the GPU will have it. WHY THIS TRIES SEVERAL TIMES RATHER THAN CALCULATING THE ANSWER. The benchmark that prompted this ran at 3.1 tokens per second with the GPU sitting at its 7 W idle draw: a CUDA-capable laptop doing every matrix multiply on its CPU. Two separate causes, and this handles the second. The first is which build was installed, and belongs to :mod:`distinct_agent.runtime`. The second is that a build with GPU support still offloads nothing unless it is told how many layers to offload, and the right number is a property of *this model on this card*: 4 GB of laptop VRAM takes every layer of a 0.6 B model and about half of a 7 B one. Guessing it needs the layer count and per-layer size from the GGUF, the free VRAM at this instant, and however much the display is about to take -- three numbers, two of which move. :func:`plan_offload` works out what fits from the card's free memory and the model's own header, and this starts there. The ladder below it is insurance, not the plan: an estimate can still be a little optimistic, so an allocation failure steps down rather than giving up, and a failure that is *not* about memory stops the ladder at once because a missing model file will not load on four layers either. The last rung is zero, which is the CPU-only behaviour this replaces, so the worst case is what used to be the only case. """ if not self.available: raise LlamaServerUnavailable(f"llama-server not found at {self.executable!r}") if not model.path.is_file(): raise RunnerUnavailable(f"model file is missing: {model.path}") # Price the load separately. It is paid once per residency and must # never be charged to a run, or queue position changes the answer. load_token = self._energy_meter.start() load_started = time.monotonic() rungs = self.gpu_layers or self._offload_ladder(model) attempts: list[str] = [] healthy = False for index, layers in enumerate(rungs): healthy = self._spawn_locked(model, layers) if healthy: self.gpu_layers_used = layers break said = self._log.tail() code = self._process.returncode if self._process is not None and self._process.poll() is not None else None attempts.append( f"-ngl {layers}: " + ("exited with %s" % code if code is not None else "timed out") ) last = index == len(rungs) - 1 self._stop_locked() if last or not _short_of_memory(said): if said: attempts[-1] += f". It said: {said}" break load_usage = self._energy_meter.stop(load_token) self.load_seconds = round(time.monotonic() - load_started, 3) self.load_energy_joules = ( float(load_usage.joules) if getattr(load_usage, "available", False) and load_usage.joules is not None else None ) self._load_charged = False if not healthy: raise LlamaServerUnavailable( "llama-server did not become healthy on " f"{LOOPBACK} ({'; '.join(attempts) or 'no attempt was made'})" ) def _offload_ladder(self, model: DiscoveredModel) -> tuple[int, ...]: """What to give this machine's card, measured once and remembered. :func:`plan_offload` says what would *fit*. That is not the same question as what is *faster*, and the difference is the whole reason this method exists: nine of thirty-two layers fitted on the benchmark laptop, started cleanly, and was slower than not using the card at all. So the settings that fit are timed, once, and the winner is written down. See :mod:`distinct_agent.offload`. """ from . import offload as _offload # noqa: PLC0415 from . import runtime as _runtime # noqa: PLC0415 - optional at import time from .gguf import read_shape # noqa: PLC0415 free, total = _runtime.accelerator_memory() if not free: return (0,) try: model_bytes = model.path.stat().st_size except OSError: return (0,) context = min(self.context_length, model.manifest.context_length) plan = plan_offload( model_bytes=model_bytes, shape=read_shape(model.path), free_vram_bytes=free, context_tokens=context, ) if plan == (0,): return plan key = _offload.machine_key( model_id=model.manifest.id, gpu=_runtime.accelerator_name(), vram_bytes=total, build=Path(self.executable).parent.name, context=context, ) known = _offload.remembered(key) if known is None: measured = self._time_offloads(model, _offload.candidates(plan)) known = _offload.choose(measured) _offload.remember(key, known, measured) self.offload_measurements = measured return (known, 0) if known else (0,) def _time_offloads( self, model: DiscoveredModel, settings: tuple[int, ...] ) -> tuple[Any, ...]: """Start the server at each setting and time one short generation. Runs before the model is resident for real, so nothing it does is charged to a job. It costs one model load per setting, once per machine and model, and it is the only thing here that can tell the difference between an offload that helps and one that does not. """ from .offload import Measurement # noqa: PLC0415 results: list[Any] = [] for layers in settings: if self._notify is not None: self._notify(f"Timing {layers} layer(s) on the GPU") healthy = self._spawn_locked(model, layers) if not healthy: results.append(Measurement(layers, 0.0, "did not start")) self._stop_locked() continue try: body, mode = self._generate( self.PROBE_PROMPT, max_tokens=self.PROBE_TOKENS, temperature=0.0, timeout=self.PROBE_TIMEOUT, ) speed = float((body.get("timings") or {}).get("predicted_per_second") or 0.0) results.append(Measurement(layers, speed)) except Exception as exc: # noqa: BLE001 - a setting that fails is a result results.append(Measurement(layers, 0.0, f"{type(exc).__name__}")) finally: self._stop_locked() return tuple(results) def _spawn_locked(self, model: DiscoveredModel, gpu_layers: int) -> bool: """Start one attempt and say whether it came up. Never raises.""" port = _free_port() command = [ self.executable, "-m", str(model.path), "--host", LOOPBACK, "--port", str(port), "-c", str(min(self.context_length, model.manifest.context_length)), "--no-webui", # A build without GPU support prints one warning and carries on, # so this is safe to pass unconditionally and saves the worker # having to know which build it installed. "-ngl", str(gpu_layers), ] # A CRASHED RUNNER USED TO TAKE ITS REASON WITH IT. # # Both streams went to DEVNULL, so when llama-server died mid-request # the operator got "llama-server became unreachable: [WinError 10054]" # and nothing else -- not the out-of-memory line, not the context # overflow, not the unsupported-flag complaint that had made it exit # during start-up. The one component whose failures a volunteer cannot # debug from the outside was the one component whose output was # discarded. # # Its last output is now kept in a bounded ring and attached to the # error. Bounded because llama-server is chatty per token and an # unbounded pipe on a long residency is a slow leak; the tail is the # part that says why it stopped. self._log = _ProcessLog() process, isolation, guard = spawn_restricted( command, self.restriction, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, shell=False, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), env=build_llama_environment(), ) self._log.follow(process.stderr) self._process, self._guard, self._port = process, guard, port self._model_path, self.isolation = model.path, isolation self._model_id = model.manifest.id return self._wait_for_health() def _wait_for_health(self) -> bool: assert self._process is not None deadline = self._clock() + self.startup_timeout_seconds url = f"http://{LOOPBACK}:{self._port}/health" while self._clock() < deadline: if self._process.poll() is not None: return False try: with urllib.request.urlopen(url, timeout=3) as response: if response.status == 200: return True except (urllib.error.URLError, OSError): time.sleep(0.5) return False # Generation ----------------------------------------------------- #: Chat first, raw completion only if the server has no chat endpoint. CHAT_PATH = "/v1/chat/completions" COMPLETION_PATH = "/completion" PROMPT_MODE_CHAT = "chat-template" PROMPT_MODE_RAW = "raw-completion" def _generate( self, prompt: str, *, max_tokens: int, temperature: float, timeout: float, schema: Mapping[str, Any] | None = None, ) -> tuple[Mapping[str, Any], str]: """Ask the model, through its own chat template wherever possible. THIS IS THE DIFFERENCE BETWEEN AN ANSWER AND A CONTINUATION. llama.cpp's ``/completion`` endpoint is raw text completion: whatever you send is continued, verbatim, with no template applied. Every model in this catalogue is an *instruct* model, which means it was trained to respond to a specific framing (for Qwen3, ``<|im_start|>user`` ... ``<|im_end|><|im_start|>assistant``) that lives in the GGUF's own chat template. Send it bare prose and it does exactly what a base language model does: it continues the document. That is not a hypothetical. Asked "In one sentence, why can a smaller language model use less energy per answer?" through ``/completion``, Qwen3-0.6B replied: A) Because it has more neurons. B) Because it has fewer neurons. C) Because it has more connections. D) Because it has fewer connections. Answer: B **Process:** 1. Identify the key elements... It had not misunderstood the question. It was never asked one. It saw the opening of a quiz and wrote the rest of the quiz. ``/v1/chat/completions`` hands the message to llama-server as a *turn*, and llama-server applies the template the model was published with. The same model then answers. The raw endpoint remains as a fallback for a server too old to have the chat route, and when it is used the run says so in ``prompt_mode``, because an answer produced without a template is a different artefact and a run log that could not distinguish them would be lying by omission. """ # ASKING NICELY FOR JSON DOES NOT WORK ON A SEVEN BILLION PARAMETER # MODEL, AND IT DOES NOT HAVE TO. # # The harness needs one JSON object per turn. The prompt said so in # capitals, gave a worked example, and re-asked once when prose came # back; the model still wrote a perfectly good paragraph *about* the # document it had been asked to create, called nothing, and produced # no file. Eleven of fourteen benchmark failures were that. # # llama.cpp can constrain the sampler to a JSON schema, which turns # "please reply in this shape" from an instruction the model may # ignore into a shape it cannot leave. A build that does not support # it says so with a 400, and the flag below stops it being asked # again, so this degrades to the behaviour it replaces rather than # failing. constrain = schema is not None and self.supports_schema is not False chat_payload = { "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": temperature, "stream": False, # Qwen3 and the other hybrid reasoning models interleave a # block by default. It is not the answer, the user did not # ask for it, and it consumes the output budget. Asking the # template to omit it is the clean route; _answer_text strips any # block that arrives anyway, because not every build honours this. "chat_template_kwargs": {"enable_thinking": False}, } if constrain: chat_payload["response_format"] = { "type": "json_schema", "json_schema": {"name": "action", "strict": True, "schema": dict(schema or {})}, } try: body = self._post(self.CHAT_PATH, chat_payload, timeout) if constrain: self.supports_schema = True return body, self.PROMPT_MODE_CHAT except RunnerTimedOut: raise except _NoSuchEndpoint: pass except RunnerError as refused: if not constrain: raise # Refused with a schema attached. Assume the schema is why, and # carry on without it rather than losing the run -- but SAY SO. # # This was silent, and silence cost an evening. A build that # refuses the grammar falls back to asking the model nicely, which # is the behaviour every one of these fixes exists to replace, and # from the outside the run looks identical to one where the # constraint worked and the model ignored it. Those need different # fixes, so they have to be distinguishable. self.supports_schema = False self.schema_refusal = str(refused)[:400] if self._notify is not None: self._notify( "llama-server refused the action schema, so the model is " f"only being asked to follow it: {self.schema_refusal}" ) return self._generate( prompt, max_tokens=max_tokens, temperature=temperature, timeout=timeout, schema=None, ) raw_payload = { "prompt": prompt, "n_predict": max_tokens, "temperature": temperature, "stream": False, "cache_prompt": True, } if constrain: raw_payload["json_schema"] = dict(schema or {}) return self._post(self.COMPLETION_PATH, raw_payload, timeout), self.PROMPT_MODE_RAW def _post( self, path: str, payload: Mapping[str, Any], timeout: float ) -> Mapping[str, Any]: request = urllib.request.Request( f"{self.base_url}{path}", data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(request, timeout=timeout) as response: body = json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: if exc.code in {404, 501}: raise _NoSuchEndpoint(path) from exc # THE BODY IS THE DIAGNOSIS. llama-server refuses a request that # does not fit the context window with a 400 and a sentence saying # so by name. Reporting only "HTTP 400" turned the one failure a # volunteer can actually fix -- shorten the transcript -- into an # anonymous number. raise RunnerError( f"llama-server rejected the request: HTTP {exc.code}{_http_detail(exc)}" ) from exc except TimeoutError as exc: raise RunnerTimedOut(f"inference exceeded {timeout:g} seconds") from exc except (urllib.error.URLError, OSError) as exc: # Attach what it said on the way out. "became unreachable" names # the symptom; the tail names the cause, and an operator can only # act on the second. said = self._log.tail() if getattr(self, "_log", None) else "" gone = self._process is not None and self._process.poll() is not None detail = f"llama-server became unreachable: {exc}" if gone: detail += f" (it exited with {self._process.returncode})" if said: detail += f". It said: {said}" raise _TransportLost(detail) from exc if not isinstance(body, Mapping): raise RunnerError("llama-server returned a malformed response") return body def stop(self) -> None: with self._lock: self._stop_locked() def _stop_locked(self) -> None: process, guard = self._process, self._guard self._process = self._guard = self._port = self._model_path = None self._model_id = None self.load_energy_joules = self.load_seconds = None self.gpu_layers_used = None self._load_charged = False if process is not None and process.poll() is None: process.terminate() try: process.communicate(timeout=10) except subprocess.TimeoutExpired: process.kill() if guard is not None: guard.close() def __enter__(self) -> LlamaServerRunner: return self def __exit__(self, *exc: object) -> None: self.stop() # -- inference ---------------------------------------------------- def run( self, model: DiscoveredModel, prompt: str, *, cancel_event: threading.Event | None = None, progress: Callable[[float, str], None] | None = None, limits: Mapping[str, Any] | None = None, ) -> InferenceResult: if not isinstance(prompt, str) or not prompt: raise RunnerError("prompt must be non-empty text") limits = limits or {} max_tokens = int(limits.get("max_output_tokens", 512)) timeout = float(limits.get("timeout_seconds", 600.0)) was_resident = self.resident_model_id == model.manifest.id if progress is not None: from .harness import emit emit( progress, 0.05, "Model already resident" if was_resident else "Loading the model", phase=PHASE_LOADING_MODEL, step={ "kind": STEP_PHASE, "text": ( f"{model.manifest.id} is already loaded" if was_resident else f"Loading {model.manifest.id} into memory" ), }, ) self.ensure_started(model) # Only the run that actually caused the load hears about it, and it # hears about it as a separate figure rather than inside its own. triggered_load = not was_resident and not self._load_charged if triggered_load: self._load_charged = True if cancel_event is not None and cancel_event.is_set(): raise RunnerCancelled("cancelled before inference started") if progress is not None: from .harness import emit emit(progress, 0.2, "Generating", phase=PHASE_GENERATING) started = time.monotonic() temperature = float(limits.get("temperature", 0.2)) # Carried in limits rather than in the signature, so a runner that # cannot constrain its output ignores it instead of refusing the call. schema = limits.get("response_schema") schema = schema if isinstance(schema, Mapping) else None try: body, prompt_mode = self._generate( prompt, max_tokens=max_tokens, temperature=temperature, timeout=timeout, schema=schema, ) except _TransportLost as lost: body, prompt_mode = self._retry_after_transport_loss( model, prompt, lost, max_tokens=max_tokens, temperature=temperature, timeout=timeout, schema=schema, ) except RunnerTimedOut: # LET IT FINISH TIDYING UP BEFORE THE NEXT RUN ARRIVES. # # A client-side timeout closes the socket mid-generation. # llama-server notices, logs `srv stop: cancel task`, and unwinds # the slot -- which takes a moment, during which it will reset a # new connection. That reset landed on whichever run came next, so # a single slow workload failed two: itself, and the innocent one # behind it. Waiting here means the run that caused the timeout is # the run that pays for it. self._await_health(self.recovery_seconds) raise text = _answer_text(body, prompt_mode) if not text: raise RunnerError("llama-server returned an empty completion") if progress is not None: progress(1.0, "Local inference complete") timings = body.get("timings") or {} return InferenceResult( text=text, usage={ "runner": self.name, "input_characters": len(prompt), "output_characters": len(text), # Real token counts, which the CLI could not report. "input_tokens": _token_count(body, prompt_mode, "prompt"), "output_tokens": _token_count(body, prompt_mode, "completion"), "tokens_per_second": timings.get("predicted_per_second"), "stop_reason": _stop_reason(body, prompt_mode), # Which endpoint answered. A raw completion is a materially # different thing from a templated chat turn, and a run log has # to be able to say which one it got. "prompt_mode": prompt_mode, # Whether the sampler was held to the action schema. An answer # produced under a constraint is a different artefact from one # produced freely, in the same way a templated turn is. "schema_constrained": bool(schema) and self.supports_schema is not False, "max_output_tokens": max_tokens, "elapsed_seconds": round(time.monotonic() - started, 3), "isolation": self.isolation.to_dict() if self.isolation else None, # State the boundary, exactly as scope='cpu-package' is stated. # The caller's meter brackets this call only, so the figure is # inference alone whether the model was already resident or # loaded a moment ago. Cold and warm runs are comparable. "energy_boundary": ENERGY_BOUNDARY_INFERENCE_ONLY, "model_was_resident": was_resident, "triggered_model_load": triggered_load, # Reported beside the run, never inside it. None means the load # was not measured, which is not the same as a load costing # nothing. "model_load_joules": self.load_energy_joules if triggered_load else None, "model_load_seconds": self.load_seconds if triggered_load else None, # Whether this answer came off the GPU, and how much of it did. # On a project whose figures are joules per run, "which device # did the arithmetic" is not a diagnostic detail, it is the # single largest term in the number being reported. "gpu_layers": self.gpu_layers_used, }, ) class _NoSuchEndpoint(RuntimeError): """This llama-server build does not implement the route that was tried.""" class _TransportLost(RuntimeError): """The connection to llama-server broke mid-request. Internal, and deliberately not a :class:`RunnerError`: a broken connection is the one inference failure that is often not the run's fault and is frequently survivable, so it has to be distinguishable from a refusal before anything can decide whether to try again. """ def _http_detail(exc: urllib.error.HTTPError) -> str: """The server's own words about why it said no, if it offered any.""" try: raw = exc.read().decode("utf-8", "replace") except Exception: # noqa: BLE001 - a body we cannot read is simply absent return "" if not raw: return "" try: parsed = json.loads(raw) except ValueError: parsed = None message = "" if isinstance(parsed, Mapping): error = parsed.get("error") if isinstance(error, Mapping): message = str(error.get("message") or "") elif isinstance(error, str): message = error if not message: message = str(parsed.get("message") or "") text = (message or raw).strip().replace("\n", " ") return f". It said: {text[:400]}" if text else "" #: A reasoning block, which is working-out rather than an answer. _THINK_BLOCK = re.compile(r"]*>.*?", re.DOTALL | re.IGNORECASE) #: An unterminated one, which happens when the budget ran out mid-thought. _OPEN_THINK = re.compile(r"]*>.*\Z", re.DOTALL | re.IGNORECASE) def strip_reasoning(text: str) -> str: """Remove a hybrid reasoning model's ```` working-out. The block is not the answer. Leaving it in puts several hundred tokens of the model talking to itself at the top of a reply that was asked for in one sentence, which is the single loudest way an interface can look broken. An *unterminated* block is removed too, and that case is why this is more than a tidy-up: when the output budget runs out inside the thought, the entire reply is working-out with no answer after it. Stripping it leaves an empty string, and an empty completion is refused upstream with a reason, which is a far better outcome than presenting a monologue as an answer. """ cleaned = _THINK_BLOCK.sub("", text) cleaned = _OPEN_THINK.sub("", cleaned) return cleaned.strip() def _answer_text(body: Mapping[str, Any], prompt_mode: str) -> str: """The assistant's reply, from whichever response shape arrived.""" if prompt_mode == LlamaServerRunner.PROMPT_MODE_CHAT: choices = body.get("choices") if not isinstance(choices, (list, tuple)) or not choices: return "" first = choices[0] if isinstance(choices[0], Mapping) else {} message = first.get("message") if isinstance(first.get("message"), Mapping) else {} content = message.get("content") return strip_reasoning(content) if isinstance(content, str) else "" content = body.get("content") return strip_reasoning(content) if isinstance(content, str) else "" def _token_count(body: Mapping[str, Any], prompt_mode: str, which: str) -> Any: if prompt_mode == LlamaServerRunner.PROMPT_MODE_CHAT: usage = body.get("usage") if isinstance(usage, Mapping): return usage.get(f"{which}_tokens") return None return body.get("tokens_evaluated" if which == "prompt" else "tokens_predicted") def _stop_reason(body: Mapping[str, Any], prompt_mode: str) -> Any: if prompt_mode == LlamaServerRunner.PROMPT_MODE_CHAT: choices = body.get("choices") if isinstance(choices, (list, tuple)) and choices and isinstance(choices[0], Mapping): return choices[0].get("finish_reason") return None return body.get("stop_type")