diff --git a/services/model-gateway/throughput_callback.py b/services/model-gateway/throughput_callback.py index 6706fe0..6bb4d2d 100644 --- a/services/model-gateway/throughput_callback.py +++ b/services/model-gateway/throughput_callback.py @@ -6,7 +6,7 @@ callbacks: ["throughput_callback.throughput_recorder_instance"] The dashboard's `/api/throughput/record` endpoint accepts: - {model, output_tokens_per_sec, ttft_ms, service} + {model, output_tokens_per_sec, ttft_ms, service, alias?, backend?} This callback fires after every successful completion and POSTs a fire-and- forget sample. Never raises into the inference path — telemetry failures are @@ -19,6 +19,7 @@ import os from datetime import datetime from typing import Any +from urllib.parse import urlparse import httpx from litellm.integrations.custom_logger import CustomLogger @@ -90,13 +91,65 @@ def _detect_service(kwargs: dict[str, Any]) -> str: return "unknown" +def _deployment_model_info(kwargs: dict[str, Any]) -> dict: + """The served deployment's model_info (including our custom `weights_file`). + + LiteLLM v1.82.3's Router._update_kwargs_with_deployment attaches it to the + request kwargs — top-level, and inside the router metadata (which the logger + later moves under litellm_params). Check every location; first hit wins.""" + for candidate in ( + kwargs.get("model_info"), + ((kwargs.get("litellm_params") or {}).get("metadata") or {}).get("model_info"), + (kwargs.get("metadata") or {}).get("model_info"), + ): + if isinstance(candidate, dict) and candidate.get("weights_file"): + return candidate + return {} + + +def _served_api_base(kwargs: dict[str, Any]) -> str: + slo = kwargs.get("standard_logging_object") or {} + return str( + slo.get("api_base") + or ((kwargs.get("litellm_params") or {}).get("metadata") or {}).get("api_base") + or (kwargs.get("metadata") or {}).get("api_base") + or "" + ) + + +def _detect_backend(kwargs: dict[str, Any]) -> str: + """Backend service that served the request, from the deployment api_base host + (e.g. 'llamacpp' vs 'llamacpp-cpu' — the failover distinction). '' if unknown.""" + try: + return (urlparse(_served_api_base(kwargs)).hostname or "")[:64] + except ValueError: + return "" + + +def _detect_alias(kwargs: dict[str, Any]) -> str: + """The model id the CALLER requested (model_group), e.g. 'local-chat'.""" + slo = kwargs.get("standard_logging_object") or {} + group = str(slo.get("model_group") or "").strip() + if group: + return group[:256] + kwarg_model = str(kwargs.get("model") or "").strip() + if "/" in kwarg_model: + kwarg_model = kwarg_model.split("/", 1)[1] + return kwarg_model[:256] + + def _resolve_model(kwargs: dict[str, Any], response_obj: Any) -> str: - """Pick the most useful model identifier for telemetry. + """Identify the model that ACTUALLY served the completion. - Prefers the underlying provider model (e.g. the actual GGUF filename - llama.cpp reports) over the LiteLLM alias, falling back to the alias if - the response doesn't carry one. Strips any `:tag` suffix. - """ + Truth source: the served deployment's model_info.weights_file — the exact GGUF, + entrypoint-substituted from the same .env the llama-server reads — so samples + key by real model, never by a routing alias like `local-chat` (which conflates + every model ever active behind it, CPU failover included). Falls back to the + legacy response/kwargs naming when the router info is missing; telemetry is + never dropped over attribution.""" + weights = str(_deployment_model_info(kwargs).get("weights_file") or "").strip() + if weights: + return weights.replace("\\", "/").rsplit("/", 1)[-1].split(":")[0][:256] resp_model = "" if isinstance(response_obj, dict): resp_model = str(response_obj.get("model") or "").strip() @@ -154,12 +207,19 @@ def _build_payload(kwargs, response_obj, start_time, end_time) -> dict[str, Any] model = _resolve_model(kwargs, response_obj) if not model: return None - return { + payload = { "model": model, "output_tokens_per_sec": round(completion_tokens / duration, 2), "service": _detect_service(kwargs), "ttft_ms": round(_ttft_ms_from_kwargs(kwargs, start_time), 1), } + alias = _detect_alias(kwargs) + backend = _detect_backend(kwargs) + if alias: + payload["alias"] = alias + if backend: + payload["backend"] = backend + return payload def _headers() -> dict[str, str]: diff --git a/services/ops-api/llamacpp_flags.py b/services/ops-api/llamacpp_flags.py index cde9d37..e90d747 100644 --- a/services/ops-api/llamacpp_flags.py +++ b/services/ops-api/llamacpp_flags.py @@ -318,3 +318,26 @@ def flag_view(effective): view["MTP_ENABLED"] = "1" if enabled else "0" view["MTP_N_MAX"] = str(n_max if n_max else 2) return view + + +# ── model-gateway coupling ──────────────────────────────────────────────────── +# .env keys the model-gateway entrypoint templates into its LiteLLM config +# (services/model-gateway/entrypoint.sh): the GGUF-derived pin-alias names + +# weights_file (LLAMACPP_MODEL), context/generation limits (CTX_SIZE, N_PREDICT), +# and the vision flag (MMPROJ). When any of these change, the gateway's advertised +# metadata is stale until it is recreated — the apply path must recreate it too. +GATEWAY_CONSUMED_KEYS = ( + "LLAMACPP_MODEL", + "LLAMACPP_CTX_SIZE", + "LLAMACPP_N_PREDICT", + "LLAMACPP_MMPROJ", +) + + +def gateway_recreate_needed(prev_effective, new_effective): + """True when any gateway-templated key's EFFECTIVE value changes (a merely + 'touched' key whose value is identical does not force a restart).""" + return any( + (prev_effective.get(k) or "") != (new_effective.get(k) or "") + for k in GATEWAY_CONSUMED_KEYS + ) diff --git a/services/ops-api/main.py b/services/ops-api/main.py index 742f29d..348d7e8 100644 --- a/services/ops-api/main.py +++ b/services/ops-api/main.py @@ -175,8 +175,9 @@ # GGUF directory (chat models + mmproj) shown in the model-config UI — the models-gguf # named volume, mounted RO at /gguf-models (compose sets LLAMACPP_MODELS_DIR to match). MODELS_DIR = Path(os.environ.get("LLAMACPP_MODELS_DIR", "/gguf-models")) -# Services that template LLAMACPP_CTX_SIZE and must also recreate when ctx changes. -MODEL_CONFIG_CTX_CONSUMERS = ["model-gateway"] +# Services that template model-config .env keys (see llamacpp_flags.GATEWAY_CONSUMED_KEYS) +# and must recreate when any of those keys' effective value changes. +MODEL_CONFIG_GATEWAY_CONSUMERS = ["model-gateway"] # Services whose GPU pin the dashboard may change. GPU_ASSIGNABLE_SERVICES = {"llamacpp", "llamacpp-embed", "comfyui", "stt", "tts"} @@ -1513,7 +1514,7 @@ async def model_config_get(_: None = Depends(verify_token)): async def model_config_post(body: ModelConfigBody, request: Request, _: None = Depends(verify_token)): """Validate + apply model-config overrides via the ONE write path: persist to - the registry, render into .env, recreate llamacpp (+ ctx consumers).""" + the registry, render into .env, recreate llamacpp (+ gateway consumers).""" errs = lf.validate_all({k: v for k, v in body.overrides.items() if v is not None}) if errs: raise HTTPException(status_code=400, detail={"validation": errs}) @@ -1528,15 +1529,13 @@ async def model_config_post(body: ModelConfigBody, request: Request, config = dict(rec.config) source_file = rec.source.get("file", "") - ctx_touched = False + prev_effective = lf.compute_effective(lf.defaults(), _read_env_values(lf.ENV_KEYS)) for k, v in body.overrides.items(): if k == "LLAMACPP_MODEL": if v: source_file = str(v) config.pop("LLAMACPP_MODEL", None) continue - if k == "LLAMACPP_CTX_SIZE": - ctx_touched = True if v is None: config.pop(k, None) else: @@ -1559,7 +1558,9 @@ async def model_config_post(body: ModelConfigBody, request: Request, rec.updated_by = "model-config" REGISTRY.upsert(rec) - services = ["llamacpp"] + (MODEL_CONFIG_CTX_CONSUMERS if ctx_touched else []) + services = ["llamacpp"] + if lf.gateway_recreate_needed(prev_effective, effective): + services += MODEL_CONFIG_GATEWAY_CONSUMERS for svc in services: _recreate_service(svc, request) diff --git a/services/v1-parity/dashboard/app.py b/services/v1-parity/dashboard/app.py index b315490..fa4c760 100644 --- a/services/v1-parity/dashboard/app.py +++ b/services/v1-parity/dashboard/app.py @@ -1432,11 +1432,25 @@ async def mcp_remove(req: McpRemoveRequest): # --- Token Throughput --- -# In-memory store: model -> list of output_tokens_per_sec (rolling, max 500) -_throughput_samples: dict[str, list[float]] = {} -_ttft_samples: dict[str, list[float]] = {} +_throughput_samples: dict[str, list[dict]] = {} # {"tps": float, "ts": epoch} +_ttft_samples: dict[str, list[dict]] = {} # {"ms": float, "ts": epoch} _MAX_SAMPLES_PER_MODEL = 500 _MAX_TRACKED_MODELS = 50 +# v2: samples are timestamped dicts. v1 stored bare floats keyed largely by routing +# ALIAS (one `local-chat` bucket conflating every model ever active behind it, CPU +# failover included) — unattributable, so version bumps trigger a clean reset. +_THROUGHPUT_STORE_VERSION = 2 +_SAMPLE_MAX_AGE_SEC = 7 * 86400 # models with no sample in 7 days leave the store + + +def _evict_stale_models(now: float) -> None: + """Drop models whose newest sample is older than _SAMPLE_MAX_AGE_SEC. + Call while holding _state_lock.""" + cutoff = now - _SAMPLE_MAX_AGE_SEC + for store in (_throughput_samples, _ttft_samples): + stale = [m for m, s in store.items() if not s or s[-1]["ts"] < cutoff] + for m in stale: + del store[m] # Last benchmark result (persists across page refresh until dashboard restart) _last_benchmark: dict | None = None @@ -1451,15 +1465,30 @@ async def mcp_remove(req: McpRemoveRequest): def _load_throughput_state() -> None: - """Load throughput samples and last benchmark from disk (R4).""" + """Load throughput samples and last benchmark from disk (R4). v1 files (no + version field) get a clean reset — their samples are un-timestamped and + alias-conflated; only last_benchmark carries over.""" global _throughput_samples, _ttft_samples, _last_benchmark, _service_usage if not _THROUGHPUT_FILE.exists(): return try: data = json.loads(_THROUGHPUT_FILE.read_text(encoding="utf-8")) - _throughput_samples = {k: v for k, v in (data.get("samples") or {}).items() if isinstance(v, list)} - _ttft_samples = {k: v for k, v in (data.get("ttft_samples") or {}).items() if isinstance(v, list)} _last_benchmark = data.get("last_benchmark") if isinstance(data.get("last_benchmark"), dict) else None + if data.get("version") != _THROUGHPUT_STORE_VERSION: + logger.warning( + "Throughput store is v%s (want v%s) — resetting samples, keeping last_benchmark", + data.get("version", 1), _THROUGHPUT_STORE_VERSION, + ) + _save_throughput_state() + return + _throughput_samples = { + k: [s for s in v if isinstance(s, dict) and "tps" in s and "ts" in s] + for k, v in (data.get("samples") or {}).items() if isinstance(v, list) + } + _ttft_samples = { + k: [s for s in v if isinstance(s, dict) and "ms" in s and "ts" in s] + for k, v in (data.get("ttft_samples") or {}).items() if isinstance(v, list) + } _service_usage = [u for u in (data.get("service_usage") or []) if isinstance(u, dict)][-_MAX_SERVICE_USAGE:] except Exception as e: logger.warning("Throughput state load failed: %s", e) @@ -1471,6 +1500,7 @@ def _save_throughput_state() -> None: _THROUGHPUT_FILE.parent.mkdir(parents=True, exist_ok=True) tmp = _THROUGHPUT_FILE.with_suffix(".json.tmp") tmp.write_text(json.dumps({ + "version": _THROUGHPUT_STORE_VERSION, "samples": _throughput_samples, "ttft_samples": _ttft_samples, "last_benchmark": _last_benchmark, @@ -1516,6 +1546,8 @@ class ThroughputRecordRequest(BaseModel): output_tokens_per_sec: float = Field(default=0.0, ge=0, le=1e6) service: str = Field(default="", max_length=64) ttft_ms: float = Field(default=0.0, ge=0, le=1e6) + alias: str = Field(default="", max_length=256) + backend: str = Field(default="", max_length=64) @app.post("/api/throughput/record") @@ -1524,18 +1556,20 @@ async def throughput_record(req: ThroughputRecordRequest): model = req.model.strip() if not model or req.output_tokens_per_sec <= 0: return {"ok": True} + now = time.time() with _state_lock: + _evict_stale_models(now) if model not in _throughput_samples: if len(_throughput_samples) >= _MAX_TRACKED_MODELS: return {"ok": True} _throughput_samples[model] = [] - _throughput_samples[model].append(req.output_tokens_per_sec) + _throughput_samples[model].append({"tps": req.output_tokens_per_sec, "ts": now}) if len(_throughput_samples[model]) > _MAX_SAMPLES_PER_MODEL: _throughput_samples[model] = _throughput_samples[model][-_MAX_SAMPLES_PER_MODEL:] if req.ttft_ms > 0 and (model in _ttft_samples or len(_ttft_samples) < _MAX_TRACKED_MODELS): if model not in _ttft_samples: _ttft_samples[model] = [] - _ttft_samples[model].append(req.ttft_ms) + _ttft_samples[model].append({"ms": req.ttft_ms, "ts": now}) if len(_ttft_samples[model]) > _MAX_SAMPLES_PER_MODEL: _ttft_samples[model] = _ttft_samples[model][-_MAX_SAMPLES_PER_MODEL:] # Service usage (which service is taxing which model) @@ -1543,9 +1577,11 @@ async def throughput_record(req: ThroughputRecordRequest): _service_usage.append({ "model": model, "service": service, + "alias": req.alias.strip()[:256], + "backend": req.backend.strip()[:64], "tps": round(req.output_tokens_per_sec, 1), "ttft_ms": round(req.ttft_ms, 1) if req.ttft_ms > 0 else 0.0, - "ts": time.time(), + "ts": now, }) if len(_service_usage) > _MAX_SERVICE_USAGE: _service_usage[:] = _service_usage[-_MAX_SERVICE_USAGE:] @@ -1594,31 +1630,86 @@ async def throughput_service_usage(): return {"by_model": result, "ok": True} +# Authoritative active model, from the same ops-controller /model-config the Model +# Control tab uses. Cached (positive AND negative) so a 10s-poll dashboard doesn't +# hammer ops; control_plane_ok=False means "unreachable" and the UI says so instead +# of guessing. Reachable-but-unconfigured (file=None, ok=True) is NOT an error. +_ACTIVE_MODEL_CACHE_TTL = 30.0 +_active_model_cache: dict = {"checked": 0.0, "value": None} +_active_model_fetch_lock = asyncio.Lock() + + +def _gateway_pin_alias(gguf: str | None) -> str | None: + """Gateway pin-alias for a GGUF — same derivation as the model-gateway + entrypoint (basename, .gguf stripped, lowercased). Derived HERE, in one + place, so no UI re-implements it.""" + if not gguf: + return None + name = gguf.replace("\\", "/").rsplit("/", 1)[-1].lower() + return name.removesuffix(".gguf") + + +async def _throughput_active_model() -> dict: + """Return {"ok": bool, "file": str | None} — ok reflects control-plane + reachability (HTTP 200 with a dict body), file is the active GGUF or None + when reachable but unconfigured. Cached (positive AND negative).""" + now = time.monotonic() + if now - _active_model_cache["checked"] < _ACTIVE_MODEL_CACHE_TTL: + return _active_model_cache["value"] + async with _active_model_fetch_lock: + # Re-check: a concurrent caller may have refreshed while we waited. + now = time.monotonic() + if now - _active_model_cache["checked"] < _ACTIVE_MODEL_CACHE_TTL: + return _active_model_cache["value"] + code, data = await _ops_request("GET", "/model-config", timeout=3.0) + if code == 200 and isinstance(data, dict): + value = {"ok": True, "file": str(data["active_model"]) if data.get("active_model") else None} + else: + logger.warning("throughput active-model fetch failed (HTTP %s)", code) + value = {"ok": False, "file": None} + _active_model_cache["checked"] = now + _active_model_cache["value"] = value + return value + + @app.get("/api/throughput/stats") async def throughput_stats(): - """Return per-model throughput stats: peak, p50, p95, p99, latest, sample_count. Includes last_benchmark if available.""" + """Per-model throughput stats over timestamped samples: peak, p50/p95/p99, latest, + sample_count, first_ts/last_ts. Includes last_benchmark if available.""" result: dict[str, dict] = {} + now = time.time() with _state_lock: + _evict_stale_models(now) snapshot = {m: list(s) for m, s in _throughput_samples.items()} ttft_snapshot = {m: list(s) for m, s in _ttft_samples.items()} benchmark = dict(_last_benchmark) if _last_benchmark else None for model, samples in snapshot.items(): if not samples: continue - sorted_s = sorted(samples) - ttfts = ttft_snapshot.get(model, []) + tps_vals = [s["tps"] for s in samples] + sorted_s = sorted(tps_vals) + ttfts = [s["ms"] for s in ttft_snapshot.get(model, [])] sorted_ttfts = sorted(ttfts) result[model] = { - "latest": round(samples[-1], 1), - "peak": round(max(samples), 1), + "latest": round(tps_vals[-1], 1), + "peak": round(max(tps_vals), 1), "p50": round(_percentile(sorted_s, 50), 1), "p95": round(_percentile(sorted_s, 95), 1), "p99": round(_percentile(sorted_s, 99), 1), "ttft_p50_ms": round(_percentile(sorted_ttfts, 50), 1) if sorted_ttfts else 0.0, "ttft_p95_ms": round(_percentile(sorted_ttfts, 95), 1) if sorted_ttfts else 0.0, "sample_count": len(samples), + "first_ts": samples[0]["ts"], + "last_ts": samples[-1]["ts"], } - out: dict = {"models": result, "ok": True} + active = await _throughput_active_model() + out: dict = { + "models": result, + "ok": True, + "active_model": active["file"], + "active_model_alias": _gateway_pin_alias(active["file"]), + "control_plane_ok": active["ok"], + } if benchmark: out["last_benchmark"] = benchmark return out @@ -1638,20 +1729,22 @@ async def performance_summary(): for model, samples in snapshot.items(): if not samples: continue - sorted_s = sorted(samples) - ttfts = ttft_snapshot.get(model, []) + tps_vals = [s["tps"] for s in samples] + sorted_s = sorted(tps_vals) + ttfts = [s["ms"] for s in ttft_snapshot.get(model, [])] sorted_ttfts = sorted(ttfts) top_models.append( { "model": model, - "latest_tps": round(samples[-1], 1), + "latest_tps": round(tps_vals[-1], 1), "p95_tps": round(_percentile(sorted_s, 95), 1), "latest_ttft_ms": round(ttfts[-1], 1) if ttfts else 0.0, "p95_ttft_ms": round(_percentile(sorted_ttfts, 95), 1) if sorted_ttfts else 0.0, "sample_count": len(samples), + "last_ts": samples[-1]["ts"], } ) - top_models.sort(key=lambda item: item["sample_count"], reverse=True) + top_models.sort(key=lambda item: item["last_ts"], reverse=True) try: rag = await asyncio.wait_for(rag_status(), timeout=2.0) except TimeoutError: @@ -1756,19 +1849,6 @@ async def throughput_benchmark(req: ThroughputBenchmarkRequest): output_tokens_per_sec = eval_count / elapsed_sec if eval_count > 0 else 0 input_tokens_per_sec = prompt_eval_count / elapsed_sec if prompt_eval_count > 0 else 0 - # Store sample for stats (peak, percentiles) - with _state_lock: - if model not in _throughput_samples: - if len(_throughput_samples) >= _MAX_TRACKED_MODELS: - pass # cap reached — skip storage but still return payload - else: - _throughput_samples[model] = [] - if model in _throughput_samples: - _throughput_samples[model].append(output_tokens_per_sec) - if len(_throughput_samples[model]) > _MAX_SAMPLES_PER_MODEL: - _throughput_samples[model] = _throughput_samples[model][-_MAX_SAMPLES_PER_MODEL:] - _maybe_save_throughput() - payload = { "ok": True, "model": model, diff --git a/services/v1-parity/dashboard/frontend/src/components/ThroughputTab.jsx b/services/v1-parity/dashboard/frontend/src/components/ThroughputTab.jsx index c333dde..58b9fb2 100644 --- a/services/v1-parity/dashboard/frontend/src/components/ThroughputTab.jsx +++ b/services/v1-parity/dashboard/frontend/src/components/ThroughputTab.jsx @@ -1,19 +1,20 @@ -// Throughput tab — inference telemetry + an on-demand benchmark. Port of the legacy -// loadPerfHero + loadThroughputServiceUsage + runThroughputBenchmark. Consumes: -// - GET /api/throughput/stats — per-model tok/s percentiles + last_benchmark +// Throughput tab — inference telemetry + an on-demand benchmark. Consumes: +// - GET /api/throughput/stats — per-model tok/s percentiles (timestamped, +// 7-day retention) + the AUTHORITATIVE active_model (ops /model-config), its +// server-derived active_model_alias (gateway pin-alias), control_plane_ok +// (ops reachability, distinct from an empty/no-model config), + last_benchmark // - GET /api/throughput/service-usage — which service drove which model, recent tok/s // - GET /api/performance/summary — context size + fleet summary -// - GET /api/llm/models — to pick the benchmark target model // - POST /api/throughput/benchmark — run a quick tok/s benchmark (confirm + pending) -// Telemetry polls every 10s (paused when hidden); the benchmark is a manual action that -// disables the button while running and refreshes the telemetry when it lands. +// The hero is the registry's active model — never inferred from sample recency or +// counts. Sample keys are real GGUF basenames (the gateway callback attributes each +// completion to the deployment that served it). Telemetry polls every 10s (paused when +// hidden); the benchmark targets the active model's gateway pin-alias (from the server, +// not re-derived client-side). import { useCallback, useState } from 'react' import { api, usePolling } from '../api.js' import { useToast } from './Toast.jsx' -const EMBED_RE = /embed|bge|mxbai|arctic-embed|granite-embedding|paraphrase-multilingual/ -const isEmbeddingModel = (name) => EMBED_RE.test((name || '').toLowerCase()) - // Callers that reach the gateway with the stock OpenAI SDK send its class name as the "service". // Render those as an honest "unidentified caller" instead of dressing a raw SDK class up as a // curated service identity (the real fix is server-side attribution; this stops the UI lying). @@ -66,24 +67,35 @@ export default function ThroughputTab() { const toast = useToast() const { data, error, refresh } = usePolling(async () => { - const [stats, usage, summary, llm] = await Promise.all([ + const [stats, usage, summary] = await Promise.all([ api.get('/api/throughput/stats').catch(() => ({ ok: false, models: {} })), api.get('/api/throughput/service-usage').catch(() => ({ ok: false, by_model: {} })), api.get('/api/performance/summary').catch(() => null), - api.get('/api/llm/models').catch(() => ({ models: [] })), ]) - return { stats, usage, summary, llm } + return { stats, usage, summary } }, 10000) const stats = data?.stats const summary = data?.summary - const llms = (data?.llm?.models || []).filter((m) => !isEmbeddingModel(m.name)) - const models = stats?.ok && stats.models ? stats.models : {} - const entries = Object.entries(models) + // statsOk: did OUR /api/throughput/stats call succeed. controlPlaneOk: given statsOk, + // did the ops-controller /model-config lookup behind it succeed. These are distinct + // failure modes — conflating them blames the control plane for a dashboard-API outage. + const statsOk = stats?.ok === true + const controlPlaneOk = statsOk ? (stats.control_plane_ok !== false) : true + const models = statsOk && stats.models ? stats.models : {} + + // Authoritative model state — ops-controller /model-config via /stats. Never guessed: + // null means either the control plane is unreachable or reachable-but-unconfigured; + // the hero panel below distinguishes those (see controlPlaneOk). + const activeModel = statsOk ? (stats.active_model ?? null) : null + const activeStats = activeModel ? models[activeModel] : null + + // Other models with recent samples (the store evicts after 7 days) — history, not "Active". + const historyRows = Object.entries(models) + .filter(([name]) => name !== activeModel) + .sort((a, b) => (b[1].last_ts || 0) - (a[1].last_ts || 0)) - // Service-usage rows carry REAL per-service timestamps (unlike /stats, which has no age - // eviction). Compute them first so the hero can be picked by recency, not raw sample count. const byModel = data?.usage?.ok ? (data.usage.by_model || {}) : {} const usageRows = [] Object.entries(byModel).forEach(([model, info]) => { @@ -92,23 +104,6 @@ export default function ThroughputTab() { usageRows.sort((a, b) => (b.svc.last_ts || 0) - (a.svc.last_ts || 0)) const maxTps = Math.max(1, ...usageRows.map((r) => r.svc.last_tps || 0)) - // Hero = the model that served most RECENTLY (real timestamps), NOT the most-sampled one: - // /stats never evicts, so a retired model keeps the top sample_count forever and would be - // mislabeled "Active". Fall back to sample_count only when there's no timestamped usage. - const recentTsByModel = {} - for (const { model, svc } of usageRows) { - if (svc.last_ts) recentTsByModel[model] = Math.max(recentTsByModel[model] || 0, svc.last_ts) - } - const mostRecent = Object.entries(recentTsByModel).sort((a, b) => b[1] - a[1])[0] - let hero = null - if (mostRecent && models[mostRecent[0]]) { - hero = [mostRecent[0], models[mostRecent[0]], mostRecent[1]] - } else { - const s = [...entries].sort((a, b) => (b[1].sample_count || 0) - (a[1].sample_count || 0))[0] - hero = s ? [s[0], s[1], 0] : null - } - const heroTs = hero ? hero[2] : 0 - const ctx = summary?.llamacpp_ctx_size || 0 const ctxLabel = ctx ? (ctx >= 1000 ? `${Math.round(ctx / 1000)}K ctx` : `${ctx} ctx`) : 'no ctx' @@ -118,14 +113,17 @@ export default function ThroughputTab() { const lastBench = stats?.last_benchmark || null const shownBench = result || lastBench + // Gateway pin-alias for the active GGUF, derived server-side (single source of truth — + // see _gateway_pin_alias in app.py). Pin (not local-chat) so the benchmark measures the + // GPU deployment and honestly errors if it's evicted, instead of silently measuring the + // CPU fallback. Falls back to local-chat only when there's no server-derived alias. + const benchTarget = (statsOk && stats.active_model_alias) || 'local-chat' + const runBenchmark = useCallback(async () => { - // First non-embedding LLM, else the always-valid `local-chat` alias — never the hero, which - // can be a stale/retired model (see the recency note above); benchmarking the wrong model silently. - const model = llms[0]?.name || 'local-chat' - if (!confirm(`Run a throughput benchmark on "${model}"?\n\nThis sends a short generation through the Model Gateway and reports tok/s.`)) return + if (!confirm(`Run a throughput benchmark on "${benchTarget}"?\n\nThis sends a short generation through the Model Gateway and reports tok/s.`)) return setRunning(true) try { - const d = await api.post('/api/throughput/benchmark', { model }) + const d = await api.post('/api/throughput/benchmark', { model: benchTarget }) setResult(d) toast(`Benchmark: ${d.output_tokens_per_sec} tok/s (${d.model})`, 'success') refresh() @@ -134,7 +132,7 @@ export default function ThroughputTab() { } finally { setRunning(false) } - }, [llms, toast, refresh]) + }, [benchTarget, toast, refresh]) return (
@@ -154,35 +152,65 @@ export default function ThroughputTab() {
{[0, 1].map((i) =>
)}
) : ( <> - {/* Hero + percentile rail */} + {/* Active model (authoritative) + percentile rail */}
-
-
- Active model -
- {hero ? hero[0] : 'No samples yet'} -
+ {!statsOk ? ( +
+ Could not load throughput telemetry — the dashboard API request failed.
-
-
{hero ? fmt(hero[1].latest) : '—'}
-
tok/s latest
+ ) : !controlPlaneOk ? ( +
+ Model control plane unreachable — cannot determine the active model.
-
-
- {hero ? `${hero[1].sample_count || 0} samples` : '0 samples'} - · - {ctxLabel} - {hero && <>·{heroTs ? `active ${fmtAgo(heroTs)}` : 'no recent traffic'}} -
- -
- - - - - - -
+ ) : activeModel === null ? ( +
+ No active model configured — set one in Model Control. +
+ ) : ( + <> +
+
+ Active model +
+ {activeModel} +
+
+
+
+ {activeStats ? fmt(activeStats.latest) : '—'} +
+
tok/s latest
+
+
+
+ {activeStats ? `${activeStats.sample_count} samples` : 'no traffic yet — run a benchmark'} + · + {ctxLabel} + {activeStats?.last_ts && <>·last sample {fmtAgo(activeStats.last_ts)}} +
+
+ + + + + + +
+ + )} + {historyRows.length > 0 && ( +
+
Recent models (last 7 days)
+ {historyRows.map(([name, m]) => ( +
+ {name} + + p50 {fmt(m.p50)} tok/s · last seen {fmtAgo(m.last_ts)} + +
+ ))} +
+ )}
{/* Benchmark runner */} @@ -191,7 +219,7 @@ export default function ThroughputTab() {
Benchmark

- Target: {llms[0]?.name || 'local-chat'} + Target: {benchTarget}