mission-control-v2/backend/services/box_updates.py

"""Updates der KI-Box: Zwischenspeicher für den Start-Bildschirm und der Box-Adapter (Phase 2).

maintenance.updates() fragt GitHub, apt und git und braucht nach einem Neustart gern zehn Sekunden
und mehr. Der Start-Bildschirm wartet darauf nie: Er bekommt den letzten Stand (oder zunächst
nichts), frisch geholt wird im Hintergrund. Seit 24.09.2026 liegt das hier statt im Router, weil
auch der Box-Adapter (ki_box_ziel) davon liest.

ki_box_ziel() beschreibt die KI-Box im gemeinsamen Modell aus kern/ziele.py: vier Bausteine mit
Stand, Versionen und dem Knopf, der das jeweilige Update anstößt.
"""

import platform
import threading
import time

from kern.ziele import Aktion, BausteinStand, ZielStand

from services import maintenance, system, update_verlauf

_updates_lock = threading.Lock()
_updates_cache: dict = {"ts": 0.0, "daten": None, "laeuft": False, "stand": -1}
UPDATES_CACHE_S = 600

NAMEN = {"os": "Betriebssystem", "engine": "Motor (llama.cpp)", "swap": "llama-swap", "hermes": "Hermes"}
NAMEN_KURZ = {"os": "Betriebssystem", "engine": "Motor", "swap": "llama-swap", "hermes": "Hermes"}

AKTIONEN = {
    "os": Aktion("POST", "/api/maintenance/os-update",
                 "Betriebssystem-Updates jetzt einspielen? Danach wird der Stack geprüft."),
    "engine": Aktion("POST", "/api/maintenance/engine-update",
                     "Den Motor jetzt aktualisieren? Die Modelle sind dabei kurz weg; scheitert die Prüfung, "
                     "geht es zurück."),
    "swap": Aktion("POST", "/api/maintenance/swap-update",
                   "llama-swap jetzt aktualisieren? Laufende Antworten brechen ab; scheitert die Prüfung, "
                   "geht es zurück."),
    "hermes": Aktion("POST", "/api/maintenance/hermes-update",
                     "Hermes jetzt aktualisieren? Lucy ist dabei ein paar Minuten weg."),
}


def _holen() -> None:
    stand = maintenance.update_stand
    try:
        daten = maintenance.updates()
    except Exception:
        daten = None
    with _updates_lock:
        if daten is not None or _updates_cache["daten"] is None:
            _updates_cache["daten"] = daten or {}
        _updates_cache["ts"] = time.time()
        _updates_cache["stand"] = stand
        _updates_cache["laeuft"] = False


def gecacht() -> dict:
    """Letzter Stand der Update-Prüfung; ist er veraltet, wird im Hintergrund neu geholt."""
    with _updates_lock:
        # Nach jedem abgeschlossenen Update zählt maintenance.update_stand hoch — dann sofort neu holen,
        # statt bis zu 10 Minuten „Aktualisieren" für schon eingespielte Updates zu zeigen (24.09.2026).
        veraltet = (_updates_cache["daten"] is None or time.time() - _updates_cache["ts"] > UPDATES_CACHE_S
                    or _updates_cache["stand"] != maintenance.update_stand)
        if veraltet and not _updates_cache["laeuft"]:
            _updates_cache["laeuft"] = True
            threading.Thread(target=_holen, name="updates-holen", daemon=True).start()
        return _updates_cache["daten"] or {}


def gelesen() -> bool:
    """Liegt schon ein Ergebnis vor (nach einem Neustart dauert die erste Prüfung)?"""
    return _updates_cache["daten"] is not None


def bausteine(u: dict) -> list[dict]:
    """Welche Bausteine haben Neues? In der Reihenfolge, in der auch das Update läuft."""
    liste = []
    if u.get("os"):
        liste.append({"id": "os", "name": "Betriebssystem", "neu": f"{u['os']} Pakete"})
    if u.get("engine"):
        liste.append({"id": "engine", "name": "Motor", "neu": "neuer Build"})
    if u.get("swap"):
        liste.append({"id": "swap", "name": "llama-swap", "neu": "neue Version"})
    for c in u.get("components") or []:
        if c.get("key") == "hermes_agent" and c.get("update"):
            liste.append({"id": "hermes", "name": "Hermes", "neu": f"{c.get('behind', '?')} Änderungen"})
    return liste


def unklar(u: dict) -> list[dict]:
    """Bausteine, deren Update-Prüfung scheiterte — ehrlich „unbekannt" statt „aktuell"."""
    return [{"id": k, "name": NAMEN_KURZ.get(k, k), "grund": grund}
            for k, grund in (u.get("pruef_fehler") or {}).items() if grund]


# --- Box-Adapter ------------------------------------------------------------------------

def _ubuntu() -> str | None:
    try:
        info = platform.freedesktop_os_release()
    except OSError:
        return None
    return " ".join(x for x in (info.get("NAME"), info.get("VERSION_ID")) if x) or None


def _versionen(u: dict) -> dict[str, tuple[str | None, str | None]]:
    """(installiert, verfügbar) je Baustein. „Verfügbar" nur, wenn es Neues gibt — die Abfrage
    dafür ist bei GitHub zwischengespeichert (maintenance._github_json, 15 Minuten)."""
    engine = maintenance._installed_engine_build()
    swap = maintenance._installed_swap_version()
    werte: dict[str, tuple[str | None, str | None]] = {
        "os": (_ubuntu(), f"{u['os']} Pakete" if u.get("os") else None),
        "engine": (f"b{engine}" if engine else None, None),
        "swap": (f"v{swap}" if swap else None, None),
    }
    try:
        if u.get("engine") and (rel := maintenance._latest_engine_asset_release()):
            werte["engine"] = (werte["engine"][0], str(rel.get("tag_name") or "") or None)
        if u.get("swap"):
            rel = maintenance._github_json(f"repos/{maintenance.SWAP_REPO}/releases/latest")
            tag = str(rel.get("tag_name") or "") if isinstance(rel, dict) else ""
            werte["swap"] = (werte["swap"][0], tag or None)
    except Exception:
        pass
    hermes = next((c for c in u.get("components") or [] if c.get("key") == "hermes_agent"), {})
    git = system.find_hermes_agent_git() or {}
    version = maintenance._hermes_version(git["path"]) if git.get("path") else None
    werte["hermes"] = (version or hermes.get("current"),
                       f"+{hermes.get('behind')} Änderungen" if hermes.get("update") else None)
    return werte


def ki_box_ziel() -> ZielStand:
    """Die KI-Box als Ziel mit ihren vier Bausteinen."""
    u = gecacht()
    ziel = ZielStand(id="ki-box", name="KI-Box", art="ki-box", instanz="box", erreichbar=True,
                     geprueft=_updates_cache["ts"] or None,
                     rueckweg="Sicherung vor jedem Sonntags-Lauf; Motor, llama-swap und Hermes rollen "
                              "bei rotem Check selbst zurück, das Betriebssystem nicht.")
    if not gelesen():
        ziel.bausteine = [BausteinStand(id=k, name=n, zustand="wird-geprueft") for k, n in NAMEN.items()]
        return ziel
    offen = {b["id"]: b["neu"] for b in bausteine(u)}
    fehler = {b["id"]: b["grund"] for b in unklar(u)}
    fest = {p["baustein"]: p for p in update_verlauf.festgehalten()}
    versionen = _versionen(u)
    for k, name in NAMEN.items():
        installiert, verfuegbar = versionen.get(k, (None, None))
        stand = BausteinStand(id=k, name=name, zustand="aktuell", installiert=installiert, verfuegbar=verfuegbar)
        if k in fest:
            p = fest[k]
            stand.zustand, stand.grund = "festgehalten", f"Seit {p['seit']} auf {p['version']}: {p['grund']}"
        elif k in fehler:
            stand.zustand, stand.grund = "unbekannt", fehler[k]
        elif k in offen:
            stand.zustand, stand.kurz, stand.aktion = "neu", offen[k], AKTIONEN[k]
        ziel.bausteine.append(stand)
    return ziel