${escapeHtml(j.key)}
- → ${box_}
+ → ${box_}
+ ${['ready', 'error'].includes(j.phase) ? '' : ``}
${bar}
${line}
${sub ? `
${sub}
` : ''}
@@ -9862,6 +9874,21 @@
});
}
+ function deleteLlmFiles(instanceId, i) {
+ const m = _llmCatalog[i];
+ if (!m) return;
+ const b = onlineLlmBoxes().find(w => w.instanceId === instanceId);
+ const d = ((b && b.disk && b.disk.models) || []).find(x => x.repo === m.hfRepo);
+ if (!confirm(`${m.id} von ${(b && b.node) || instanceId} löschen?\n\n`
+ + `Die heruntergeladene Datei (${d ? _fmtBytes(d.bytes) : '?'}, Repo ${m.hfRepo}) wird von der Box entfernt. `
+ + `Wer es wieder will, muss es neu herunterladen.`)) return;
+ send({ action: 'llm_remove_model', targetInstance: instanceId, key: m.id, hfRepo: m.hfRepo, deleteFiles: true });
+ }
+ function cancelLlmJob(instanceId, key) {
+ if (!confirm(`Laden von ${key} abbrechen?\n\nDer halbe Download wird gelöscht, die Karte bleibt bzw. wird wie vorher.`)) return;
+ send({ action: 'llm_remove_model', targetInstance: instanceId, key, cancel: true });
+ }
+
async function provisionModel(i) {
const m = _llmCatalog[i];
if (!m) return;
diff --git a/diagnostic/server.js b/diagnostic/server.js
index a670505..1b1bd9f 100644
--- a/diagnostic/server.js
+++ b/diagnostic/server.js
@@ -874,6 +874,7 @@ function workerList() {
// aber diese Feldliste liess sie nie zum Browser durch ("meldet keine Karten").
slots: Array.isArray(w.slots) ? w.slots : null,
knownModels: w.knownModels || [],
+ disk: w.disk || null, // LLM-Box: freier Platz + heruntergeladene Modelle
online: (now - (w.last_seen || 0)) < WORKER_OFFLINE_MS,
}));
}
@@ -1443,6 +1444,7 @@ function connectRVS(forcePlain) {
// in eine Karte laden koennte.
slots: Array.isArray(p.slots) ? p.slots : (prev.slots || null),
knownModels: Array.isArray(p.knownModels) ? p.knownModels : (prev.knownModels || []),
+ disk: p.disk || prev.disk || null,
// OpenAI-Endpunkt (nur llm-Worker): Port/Bind kommen aus der .env der
// Box, der Key wird NIE uebertragen — nur ob einer gesetzt ist und ob
// die .env ihn vorgibt (dann ist das Feld hier schreibgeschuetzt).
@@ -3611,9 +3613,11 @@ wss.on("connection", (ws) => {
targetInstance: msg.targetInstance || "", gpu: String(msg.gpu ?? "") }, timestamp: Date.now() });
log("info", "llm", `Karte ${msg.gpu} leeren → ${msg.targetInstance || "?"}`);
} else if (msg.action === "llm_remove_model") {
+ // cancel = laufenden Download abbrechen, deleteFiles = Dateien von der Platte
sendToRVS_raw({ type: "llm_remove_model", payload: {
- targetInstance: msg.targetInstance || "", key: msg.key }, timestamp: Date.now() });
- log("info", "llm", `remove '${msg.key}' → ${msg.targetInstance || "?"}`);
+ targetInstance: msg.targetInstance || "", key: msg.key, hfRepo: msg.hfRepo || "",
+ cancel: !!msg.cancel, deleteFiles: !!msg.deleteFiles }, timestamp: Date.now() });
+ log("info", "llm", `${msg.cancel ? "abbrechen" : msg.deleteFiles ? "Dateien loeschen" : "remove"} '${msg.key}' → ${msg.targetInstance || "?"}`);
} else if (msg.action === "llm_test") {
// Test-Chat: kurze Nachricht direkt ans lokale LLM (llm_request/llm_response).
const reqId = "diagtest_" + Date.now();
diff --git a/xtts/llm-adapter/adapter.py b/xtts/llm-adapter/adapter.py
index 34bd765..92f5216 100644
--- a/xtts/llm-adapter/adapter.py
+++ b/xtts/llm-adapter/adapter.py
@@ -28,8 +28,11 @@ import json
import logging
import os
import re
+import hashlib
+import queue
import shlex
import shutil
+import threading
import time
import httpx
@@ -879,6 +882,7 @@ async def _announce(ws) -> None:
"models": models,
"slots": info, # pro LLM-Karte: Modell + Zustand (Diagnostic)
"knownModels": _known_models(), # was in eine Karte geladen werden KOENNTE
+ "disk": await asyncio.to_thread(_disk_info), # frei + heruntergeladene Modelle
# OpenAI-Endpunkt fuer Fremd-Tools (OpenHands & Co.). Der Key selbst wird
# NICHT gesendet (Broadcast!) — nur ob einer gesetzt ist und ob die .env
# ihn vorgibt (dann ist das Feld in der Diagnostic schreibgeschuetzt).
@@ -1127,9 +1131,12 @@ def _scan_download(oids: set) -> tuple:
inprog = fn.endswith(_INPROGRESS)
base = fn[:-len(_INPROGRESS)] if inprog else fn
try:
- size = os.path.getsize(os.path.join(root, fn))
+ st = os.stat(os.path.join(root, fn))
except OSError:
continue
+ # Der eigene Parallel-Download legt die Datei in voller Groesse an
+ # (sparse) — dann zaehlt nur, was wirklich geschrieben ist.
+ size = min(st.st_size, st.st_blocks * 512) if inprog else st.st_size
if base in oids:
done += size
found += 1
@@ -1199,6 +1206,183 @@ def _check_disk(files: list) -> tuple:
return True, note
+# Paralleler Download: llama.cpp zieht eine Datei ueber EINE Verbindung, das
+# bleibt oft bei 10-50 MB/s haengen, auch bei 1-Gbit-Leitung. Wir holen die
+# Datei vorher selbst in Stuecken ueber mehrere Verbindungen — direkt ins
+# HF-Cache-Layout, das llama.cpp selbst anlegt (blobs/
, snapshots/
+# / → Symlink, refs/main). Danach findet llama.cpp sie als
+# "liegt schon da". Klappt es nicht, laedt llama.cpp wie bisher selbst.
+DOWNLOAD_CONNECTIONS = int(os.getenv("LLM_DOWNLOAD_CONNECTIONS", "8"))
+DOWNLOAD_PIECE = 64 * 1024**2
+_provisions: dict = {} # key → {"cancel": threading.Event, "task", "phase"}
+
+
+def _repo_dir(repo: str) -> str:
+ return os.path.join(MODELS_DIR, "models--" + repo.replace("/", "--"))
+
+
+async def _hf_commit(repo: str) -> str:
+ headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
+ try:
+ async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
+ r = await c.get(f"https://huggingface.co/api/models/{repo}", headers=headers)
+ r.raise_for_status()
+ return str(r.json().get("sha") or "")
+ except Exception as e:
+ logger.warning("HF-Commit fuer %s nicht abrufbar (%s)", repo, e)
+ return ""
+
+
+def _link_snapshot(repo: str, commit: str, f: dict) -> None:
+ rdir = _repo_dir(repo)
+ link = os.path.join(rdir, "snapshots", commit, f["path"])
+ os.makedirs(os.path.dirname(link), exist_ok=True)
+ target = os.path.relpath(os.path.join(rdir, "blobs", f["oid"]), os.path.dirname(link))
+ if os.path.islink(link) or os.path.exists(link):
+ os.remove(link)
+ os.symlink(target, link)
+ os.makedirs(os.path.join(rdir, "refs"), exist_ok=True)
+ with open(os.path.join(rdir, "refs", "main"), "w") as fh:
+ fh.write(commit)
+
+
+def _download_parallel(repo: str, commit: str, f: dict, cancel: threading.Event) -> str:
+ """Eine Datei in Stuecken ueber mehrere Verbindungen. "" = fertig, sonst Grund."""
+ blobs = os.path.join(_repo_dir(repo), "blobs")
+ os.makedirs(blobs, exist_ok=True)
+ final = os.path.join(blobs, f["oid"])
+ part = final + _INPROGRESS
+ if os.path.exists(final):
+ return ""
+ if os.path.exists(part) and time.time() - os.path.getmtime(part) < 120:
+ return "es laeuft schon ein Download dieser Datei"
+ size = f["size"]
+ url = f"https://huggingface.co/{repo}/resolve/{commit}/{f['path']}"
+ headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
+ with open(part, "wb") as fh:
+ fh.truncate(size) # sparse: belegt erst beim Schreiben
+ pieces: queue.Queue = queue.Queue()
+ for start in range(0, size, DOWNLOAD_PIECE):
+ pieces.put((start, min(size, start + DOWNLOAD_PIECE) - 1))
+ errors: list = []
+
+ def worker() -> None:
+ fd = os.open(part, os.O_WRONLY)
+ try:
+ with httpx.Client(timeout=httpx.Timeout(30, read=60), follow_redirects=True,
+ headers=headers) as c:
+ while not cancel.is_set() and not errors:
+ try:
+ start, end = pieces.get_nowait()
+ except queue.Empty:
+ return
+ pos = start
+ for attempt in range(5):
+ try:
+ with c.stream("GET", url, headers={"Range": f"bytes={pos}-{end}"}) as r:
+ if r.status_code != 206:
+ raise RuntimeError(f"HTTP {r.status_code} statt Teil-Download")
+ for chunk in r.iter_bytes(1024 * 1024):
+ if cancel.is_set():
+ return
+ os.pwrite(fd, chunk, pos)
+ pos += len(chunk)
+ if pos != end + 1:
+ raise RuntimeError("Stueck unvollstaendig")
+ break
+ except Exception as e:
+ if attempt == 4 or cancel.is_set():
+ errors.append(str(e)[:200])
+ return
+ time.sleep(2 * (attempt + 1)) # weiter ab pos
+ finally:
+ os.close(fd)
+
+ threads = [threading.Thread(target=worker, daemon=True) for _ in range(DOWNLOAD_CONNECTIONS)]
+ for t in threads:
+ t.start()
+ for t in threads:
+ t.join()
+ if cancel.is_set() or errors:
+ try:
+ os.remove(part)
+ except OSError:
+ pass
+ return "abgebrochen" if cancel.is_set() else errors[0]
+ h = hashlib.sha256()
+ with open(part, "rb") as fh:
+ while chunk := fh.read(8 * 1024 * 1024):
+ if cancel.is_set():
+ os.remove(part)
+ return "abgebrochen"
+ h.update(chunk)
+ if h.hexdigest() != f["oid"]:
+ os.remove(part)
+ return "Pruefsumme stimmt nicht — Datei verworfen"
+ os.replace(part, final)
+ return ""
+
+
+async def _prefetch(ws, key: str, repo: str, files: list, card: str, cancel: threading.Event) -> str:
+ """Laedt fehlende Dateien parallel vor. "" = alles da (oder Fallback auf
+ llama.cpp), "abgebrochen" = Benutzer hat abgebrochen."""
+ files = [f for f in files if f.get("oid") and f.get("size")]
+ if DOWNLOAD_CONNECTIONS < 2 or not files:
+ return ""
+ missing = [f for f in files if not os.path.exists(os.path.join(_repo_dir(repo), "blobs", f["oid"]))]
+ if not missing:
+ return ""
+ commit = await _hf_commit(repo)
+ if not commit:
+ return ""
+ stop = asyncio.Event()
+ await _emit_progress(ws, key, "start", total=sum(f["size"] for f in files), files=len(files), gpu=card,
+ note=f"Lade mit {DOWNLOAD_CONNECTIONS} Verbindungen")
+ watcher = asyncio.create_task(_watch_download(ws, key, files, stop, final_phase=False))
+ err = ""
+ try:
+ for f in missing:
+ err = await asyncio.to_thread(_download_parallel, repo, commit, f, cancel)
+ if err:
+ break
+ if not err:
+ for f in files:
+ await asyncio.to_thread(_link_snapshot, repo, commit, f)
+ finally:
+ stop.set()
+ try:
+ await watcher
+ except Exception:
+ pass
+ if err and err != "abgebrochen":
+ logger.warning("Paralleler Download von %s fehlgeschlagen (%s) — llama.cpp laedt selbst", repo, err)
+ return ""
+ return err
+
+
+def _disk_info() -> dict:
+ """Freier Platz + fertig heruntergeladene Modelle (je Repo) fuer die Diagnostic."""
+ out = {"freeBytes": 0, "models": []}
+ try:
+ out["freeBytes"] = shutil.disk_usage(MODELS_DIR).free
+ for name in sorted(os.listdir(MODELS_DIR)):
+ if not name.startswith("models--"):
+ continue
+ blobs = os.path.join(MODELS_DIR, name, "blobs")
+ size = 0
+ for fn in (os.listdir(blobs) if os.path.isdir(blobs) else []):
+ if not fn.endswith(_INPROGRESS):
+ try:
+ size += os.path.getsize(os.path.join(blobs, fn))
+ except OSError:
+ pass
+ if size:
+ out["models"].append({"repo": name[len("models--"):].replace("--", "/", 1), "bytes": size})
+ except OSError as e:
+ logger.info("Plattenbelegung nicht lesbar: %s", e)
+ return out
+
+
async def _gpu_info(card: str) -> dict:
"""Name + VRAM einer Karte nach dem Laden (fuer 'in Karte N geladen')."""
if not card:
@@ -1222,7 +1406,7 @@ async def _emit_progress(ws, key: str, phase: str, **extra) -> None:
"phase": phase, "ts": int(time.time() * 1000), **extra})
-async def _watch_download(ws, key: str, files: list, stop: asyncio.Event) -> None:
+async def _watch_download(ws, key: str, files: list, stop: asyncio.Event, final_phase: bool = True) -> None:
"""Meldet jede Sekunde den Download-Stand, bis der Warmup fertig ist.
Ist alles da, wechselt die Phase auf 'load' (llama.cpp laedt in die Karte)."""
oids = {f["oid"] for f in files if f.get("oid")}
@@ -1249,7 +1433,7 @@ async def _watch_download(ws, key: str, files: list, stop: asyncio.Event) -> Non
# es lag schon im Cache. Im Notfallmodus (keine HF-Liste) zaehlt
# nur: es lief ein Download, und jetzt nicht mehr.
all_there = bool(oids) and found >= len(oids)
- if phase != "load" and (all_there or downloaded):
+ if final_phase and phase != "load" and (all_there or downloaded):
phase = "load"
await _emit_progress(ws, key, "load", bytes=total or b, total=total,
percent=100, cached=not downloaded)
@@ -1265,6 +1449,18 @@ async def _provision_result(ws, key: str, ok: bool, **extra) -> None:
async def _handle_provision(ws, payload: dict) -> None:
+ key = (payload.get("key") or "").strip()
+ if key in _provisions:
+ return await _provision_result(ws, key, False, gpu=str(payload.get("gpu") or ""),
+ error="Fuer dieses Modell laeuft schon ein Auftrag.")
+ _provisions[key] = {"cancel": threading.Event(), "task": asyncio.current_task(), "phase": "prep"}
+ try:
+ await _do_provision(ws, payload, _provisions[key])
+ finally:
+ _provisions.pop(key, None)
+
+
+async def _do_provision(ws, payload: dict, job: dict) -> None:
"""Laedt ein Modell in eine LLM-Karte (nur aus der Diagnostic). Ersetzt,
was dort lag. Ablauf: Registry + Belegung → Config (llama-swap laedt per
--watch-config + Preload) → Warmup, der wartet bis es bereit ist; daneben
@@ -1323,6 +1519,13 @@ async def _handle_provision(ws, payload: dict) -> None:
return await _provision_result(ws, key, False, gpu=card, error=disk_note)
if disk_note:
quant_note = "; ".join(x for x in (quant_note, disk_note) if x)
+ # Vor dem Umstellen der Karte: die alte bleibt nutzbar, solange geladen wird.
+ job["phase"] = "download"
+ pf = await _prefetch(ws, key, repo, _files_for_quant(repo_files, quant), card, job["cancel"])
+ if pf == "abgebrochen" or job["cancel"].is_set():
+ msg = "Abgebrochen — Karte unveraendert"
+ await _emit_progress(ws, key, "error", gpu=card, error=msg)
+ return await _provision_result(ws, key, False, gpu=card, error=msg)
ctx_req = int(payload.get("ctx") or 8192)
try:
cpu_moe = int(payload.get("cpuMoe") or 0)
@@ -1363,9 +1566,16 @@ async def _handle_provision(ws, payload: dict) -> None:
t0 = time.time()
# Langer Timeout: ein GGUF mit mehreren GB braucht Minuten. Mit dem normalen
# LLM_TIMEOUT_SEC meldete das frueher "Fehler", waehrend llama-swap weiterlud.
- res = await _call_llama([{"role": "user", "content": "hi"}],
- max_tokens=1, temperature=0.0, stop=None, model=key,
- timeout=WARMUP_TIMEOUT_S)
+ job["phase"] = "llama"
+ try:
+ if job["cancel"].is_set():
+ raise asyncio.CancelledError
+ res = await _call_llama([{"role": "user", "content": "hi"}],
+ max_tokens=1, temperature=0.0, stop=None, model=key,
+ timeout=WARMUP_TIMEOUT_S)
+ except asyncio.CancelledError:
+ # Abbrechen-Knopf: wie ein Fehlschlag behandeln → Karte zuruecksetzen
+ res = {"ok": False, "error": "Abgebrochen"}
dt = time.time() - t0
stop.set()
try:
@@ -1406,6 +1616,13 @@ async def _handle_provision(ws, payload: dict) -> None:
_save_registry(reg)
_generate_config()
await _reload_llama()
+ if job["cancel"].is_set():
+ # llama.cpp ist gestoppt — seinen halben Download nicht liegen lassen
+ for f in files:
+ try:
+ os.remove(os.path.join(_repo_dir(repo), "blobs", f["oid"] + _INPROGRESS))
+ except OSError:
+ pass
await _announce(ws)
note = f" — Karte {card} wieder auf {previous}" if previous else f" — Karte {card} wieder leer"
await _emit_llm_status(ws, "error", key, error=err)
@@ -1432,10 +1649,24 @@ async def _handle_unload(ws, payload: dict) -> None:
async def _handle_remove(ws, payload: dict) -> None:
"""Modell aus der Registry entfernen (und von seiner Karte, falls belegt).
- Das GGUF bleibt im Cache."""
+ Das GGUF bleibt im Cache — ausser deleteFiles. cancel = laufenden
+ Download/Ladevorgang abbrechen (gleicher Nachrichtentyp, damit die
+ RVS-Allow-List nicht erweitert werden muss)."""
key = (payload.get("key") or "").strip()
if not key:
return
+ if payload.get("cancel"):
+ job = _provisions.get(key)
+ if not job:
+ # Nichts laeuft (mehr) — haengenden Eintrag im Download-Fenster schliessen
+ return await _provision_result(ws, key, False, error="Abgebrochen (lief nicht mehr)")
+ job["cancel"].set()
+ if job["phase"] == "llama" and job.get("task"):
+ job["task"].cancel()
+ logger.info("Abbruch angefordert: %s (Phase %s)", key, job["phase"])
+ return
+ if payload.get("deleteFiles"):
+ return await _delete_model_files(ws, key, (payload.get("hfRepo") or "").strip())
_save_registry([e for e in _load_registry() if e.get("key") != key])
slots = _load_slots()
for c in [c for c, m in slots.items() if m == key]:
@@ -1448,6 +1679,35 @@ async def _handle_remove(ws, payload: dict) -> None:
logger.info("removed model %s", key)
+async def _delete_model_files(ws, key: str, repo: str) -> None:
+ """Heruntergeladene Dateien eines Repos von der Platte loeschen. Nicht,
+ solange es auf einer Karte liegt oder gerade geladen wird."""
+ reg = _load_registry()
+ repo = repo or next((e.get("hfRepo") for e in reg if e.get("key") == key), "") or ""
+ if not re.fullmatch(r"\w[\w.-]*/\w[\w.-]*", repo):
+ return await _provision_result(ws, key, False, error=f"Unbekanntes Repo '{repo}' — nichts geloescht")
+ repo_of = {e.get("key"): e.get("hfRepo") for e in reg}
+ busy = [c for c, m in _load_slots().items() if m and repo_of.get(m) == repo]
+ if busy:
+ return await _provision_result(ws, key, False, error=(
+ f"Liegt noch auf Karte {', '.join(busy)} — erst die Karte leeren oder ein anderes Modell laden."))
+ if any(repo_of.get(k) == repo for k in _provisions):
+ return await _provision_result(ws, key, False, error="Wird gerade geladen — erst abbrechen.")
+ path = _repo_dir(repo)
+ freed = sum(m["bytes"] for m in _disk_info()["models"] if m["repo"] == repo)
+ try:
+ await asyncio.to_thread(shutil.rmtree, path)
+ except FileNotFoundError:
+ pass
+ except OSError as e:
+ return await _provision_result(ws, key, False, error=f"Loeschen fehlgeschlagen: {e}")
+ _save_registry([e for e in reg if e.get("hfRepo") != repo])
+ _generate_config()
+ await _announce(ws)
+ await _provision_result(ws, key, True, removed=True, deleted=True, freedBytes=freed)
+ logger.info("Dateien von %s geloescht (%.1f GB)", repo, freed / 1024**3)
+
+
async def _handle_models_request(ws, payload: dict) -> None:
"""Erkundung fuer Clients: welche Modelle kann ich bei DIESER Box fragen?
Jede Box antwortet einzeln (RVS ist ein Broadcast-Raum) — der Client