feat(llm): Downloads parallel, abbrechbar; heruntergeladene Modelle loeschen

llm-adapter:
- Paralleler Download: llama.cpp zieht ueber EINE Verbindung (oft 1-50 MB/s,
  auch bei 1 Gbit). Die Box holt fehlende Dateien jetzt vorher selbst in
  64-MB-Stuecken ueber LLM_DOWNLOAD_CONNECTIONS (Standard 8) Verbindungen,
  prueft die sha256 und legt sie im HF-Cache-Layout ab, das llama.cpp selbst
  benutzt (blobs/<sha256>, snapshots/<commit>/<datei>, refs/main). Die Karte
  bleibt waehrenddessen beim alten Modell. Scheitert es, laedt llama.cpp wie
  bisher selbst. Fortschritt zaehlt bei sparse-Dateien nur geschriebene Bytes.
- Abbrechen: llm_remove_model mit cancel=true (kein neuer Typ → RVS-Allow-List
  unveraendert). Waehrend des Downloads: Stopp + halbe Datei weg, Karte
  unveraendert. Waehrend llama.cpp laedt: wie ein Fehlschlag → Karte zurueck.
  Laeuft nichts mehr, schliesst die Antwort den haengenden Eintrag.
- Loeschen: llm_remove_model mit deleteFiles=true entfernt das Repo aus dem
  Cache — nicht, solange es auf einer Karte liegt oder geladen wird.
- worker_hello meldet "disk": freier Platz + heruntergeladene Modelle je Repo.

Diagnostic: Abbrechen-Knopf je laufendem Download; im Katalog "auf ki-box ·
18,6 GB" mit Loeschen-Knopf und eine Zeile mit dem freien Platz je Box.

Getestet gegen HuggingFace (ggml-org/tiny-llamas, 4 Verbindungen, 2-MB-
Stuecke: Layout + Pruefsumme + Abbruch) und Loeschen inkl. Sperren.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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/<sha256>, snapshots/
+# <commit>/<datei> → 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