#!/usr/bin/env python3
# SkillFish Remote - dashboard backend daemon (Python standard library only).
# HTTPS (self-signed) + PAM login + signed session cookies. The web UI composes
# itself from the modules enabled in /etc/skillfish/dashboard.json. Runs as a
# systemd service (root) so it can authenticate via PAM and drive system actions.
# LAN-only by design; never expose to the internet without a real reverse proxy.
import os, sys, json, ssl, hmac, hashlib, base64, time, subprocess, threading, urllib.parse, re, socket, select, ipaddress
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

CONF = "/etc/skillfish/dashboard.json"
STATE = "/etc/skillfish"
SECRET_F = os.path.join(STATE, "dashboard.secret")
CERT_F = os.path.join(STATE, "dashboard-cert.pem")
KEY_F = os.path.join(STATE, "dashboard-key.pem")
WEB = "/usr/share/skillfish/dashboard"
I18N_DIR = "/usr/share/skillfish/i18n"   # il dizionario condiviso con le app
HUD = "/usr/local/bin/skillfish-hud-val"
SESSION_TTL = 8 * 3600

# Cosa e' acceso appena installato. Cinque restano spenti apposta, e per lo
# stesso motivo di fondo: fanno qualcosa che si sceglie, non che si subisce.
#   power schedule -> spegne e accende la macchina da solo;
#   game streaming -> e' ancora «in arrivo» nell'applicazione stessa;
#   AI-Ops         -> fa agire un modello sul sistema;
#   log            -> mette il giornale di sistema su una pagina web, e li'
#                     dentro passano nomi di file, indirizzi, pezzi di comandi.
#   HUD            -> disegna un pannello sopra al desktop di chi e' seduto
#                     davanti alla macchina: acceso da remoto senza che
#                     nessuno l'abbia chiesto e' invadente.
# Tutto il resto e' acceso: una dashboard che si presenta quasi vuota non fa
# capire cosa sa fare, ed era il difetto di prima (accesi solo due moduli).
MODULI_PREDEFINITI = {
    "telemetry": True,
    "status": True,
    "tuner": True,
    "ventola": True,
    "hub": True,
    "launcher": True,
    "kvm": True,
    "terminal": True,
    "ai": True,
    "rules": True,
    "zerotier": True,
    "snapshots": True,
    "giochi": True,
    "profili": True,
    "hud": False,
    "logs": False,
    "gamestream": False,
    "aiops": False,
    "wol": False,
}

DEFAULT_CONF = {"bind": "0.0.0.0", "port": 8443, "user": "skillfish",
                "reti_auto": True, "reti_extra": [],
                "modules": dict(MODULI_PREDEFINITI)}

# Human-facing module catalogue (id -> label/icon). The frontend renders cards
# only for modules that are BOTH known here AND enabled in the config.
MODULE_META = {
    "telemetry":  {"icon": "📊", "name": "Telemetria",     "name_en": "Telemetry"},
    "status":     {"icon": "🧊", "name": "Stato sistema",   "name_en": "System status"},
    "tuner":      {"icon": "🎛️", "name": "Controlli",       "name_en": "Controls"},
    "ventola":    {"icon": "🌀", "name": "Ventola",         "name_en": "Fan"},
    "snapshots":  {"icon": "🕒", "name": "Snapshot",       "name_en": "Snapshots"},
    "giochi":     {"icon": "🎮", "name": "Giochi",         "name_en": "Games"},
    "profili":    {"icon": "🎚️", "name": "Profili",        "name_en": "Profiles"},
    "hud":        {"icon": "🪟", "name": "HUD",             "name_en": "HUD"},
    "hub":        {"icon": "📦", "name": "App e pacchetti", "name_en": "Apps & packages"},
    "logs":       {"icon": "📜", "name": "Log",             "name_en": "Logs"},
    "launcher":   {"icon": "🚀", "name": "Avvio app",       "name_en": "Launcher"},
    "kvm":        {"icon": "🖥️", "name": "Desktop (KVM)",   "name_en": "Desktop (KVM)"},
    "terminal":   {"icon": "⌨️", "name": "Terminale",       "name_en": "Terminal"},
    "ai":         {"icon": "🧠", "name": "AI locale",       "name_en": "On-device AI"},
    "gamestream": {"icon": "🎮", "name": "Game streaming",  "name_en": "Game streaming"},
    "aiops":      {"icon": "🩺", "name": "AI-Ops",          "name_en": "AI-Ops"},
    "rules":      {"icon": "⚙️", "name": "Regole auto",     "name_en": "Auto rules"},
    "wol":        {"icon": "🔋", "name": "Power schedule",  "name_en": "Power schedule"},
    "zerotier":   {"icon": "🌐", "name": "ZeroTier",        "name_en": "ZeroTier"},
}


def load_conf():
    try:
        with open(CONF) as f:
            c = json.load(f)
        for k, v in DEFAULT_CONF.items():
            c.setdefault(k, v)
        return c
    except Exception:
        return dict(DEFAULT_CONF)


def get_secret():
    try:
        with open(SECRET_F, "rb") as f:
            return f.read()
    except Exception:
        s = os.urandom(32)
        try:
            os.makedirs(STATE, exist_ok=True)
            fd = os.open(SECRET_F, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
            os.write(fd, s); os.close(fd)
        except Exception:
            pass
        return s


SECRET = get_secret()


# ---------------- PAM auth (ctypes, no external deps) ----------------
import ctypes, ctypes.util

class _PamMessage(ctypes.Structure):
    _fields_ = [("msg_style", ctypes.c_int), ("msg", ctypes.c_char_p)]

class _PamResponse(ctypes.Structure):
    _fields_ = [("resp", ctypes.c_char_p), ("resp_retcode", ctypes.c_int)]

_CONV = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int,
                         ctypes.POINTER(ctypes.POINTER(_PamMessage)),
                         ctypes.POINTER(ctypes.POINTER(_PamResponse)), ctypes.c_void_p)

class _PamConv(ctypes.Structure):
    _fields_ = [("conv", _CONV), ("appdata_ptr", ctypes.c_void_p)]

try:
    _libpam = ctypes.CDLL(ctypes.util.find_library("pam"))
    _libc = ctypes.CDLL(ctypes.util.find_library("c"))
    _libc.calloc.restype = ctypes.c_void_p
    _libc.calloc.argtypes = [ctypes.c_size_t, ctypes.c_size_t]
    _libpam.pam_start.restype = ctypes.c_int
    _libpam.pam_start.argtypes = [ctypes.c_char_p, ctypes.c_char_p,
                                  ctypes.POINTER(_PamConv), ctypes.POINTER(ctypes.c_void_p)]
    _libpam.pam_authenticate.restype = ctypes.c_int
    _libpam.pam_authenticate.argtypes = [ctypes.c_void_p, ctypes.c_int]
    _libpam.pam_acct_mgmt.restype = ctypes.c_int
    _libpam.pam_acct_mgmt.argtypes = [ctypes.c_void_p, ctypes.c_int]
    _libpam.pam_end.restype = ctypes.c_int
    _libpam.pam_end.argtypes = [ctypes.c_void_p, ctypes.c_int]
    _PAM_OK = True
except Exception:
    _PAM_OK = False


# --- dizionario condiviso ----------------------------------------------------
# Le pagine web leggono le stesse traduzioni delle app native. Tenerlo in
# memoria evita di rileggere il file a ogni richiesta; il controllo del mtime
# fa vedere un aggiornamento del pacchetto senza riavviare il servizio.
_i18n_cache = {}          # lingua -> (mtime, voci)
_i18n_lock = threading.Lock()


def i18n_voci(lang):
    """Le voci della lingua chiesta, o {} se non c'e' o non si legge.

    Non solleva mai: una traduzione assente e' un fastidio, una pagina che non
    si apre e' un guasto. Stessa regola di /usr/share/skillfish/i18n.py.
    """
    if not (isinstance(lang, str) and len(lang) == 2
            and lang.isascii() and lang.isalpha() and lang.islower()):
        return {}
    # Il nome arriva dalla richiesta web e non viene usato per COMPORRE il
    # percorso: si guarda cosa c'e' nella cartella e lo si confronta. Cosi' un
    # file fuori di li' non e' vietato da un controllo che qualcuno domani puo'
    # indebolire — e' che non c'e' modo di nominarlo.
    p = None
    try:
        for nome in os.listdir(I18N_DIR):
            if nome.endswith(".json") and nome[:-5] == lang:
                p = os.path.join(I18N_DIR, nome)
                break
    except OSError:
        return {}
    if p is None:
        return {}
    try:
        m = os.path.getmtime(p)
    except OSError:
        return {}
    with _i18n_lock:
        vecchio = _i18n_cache.get(lang)
        if vecchio and vecchio[0] == m:
            return vecchio[1]
    try:
        with open(p, encoding="utf-8") as f:
            dati = json.load(f)
        voci = dati.get("voci") if isinstance(dati.get("voci"), dict) else dati
        voci = {k: v for k, v in voci.items()
                if isinstance(k, str) and isinstance(v, str) and v}
    except Exception:
        voci = {}
    with _i18n_lock:
        _i18n_cache[lang] = (m, voci)
    return voci


def pam_check(username, password, service="login"):
    if not _PAM_OK:
        return False

    @_CONV
    def conv(n_messages, messages, p_response, app_data):
        addr = _libc.calloc(n_messages, ctypes.sizeof(_PamResponse))
        p_response[0] = ctypes.cast(addr, ctypes.POINTER(_PamResponse))
        for i in range(n_messages):
            if messages[i].contents.msg_style == 1:  # PAM_PROMPT_ECHO_OFF
                pw = password.encode() + b"\x00"
                dst = _libc.calloc(len(pw), 1)
                ctypes.memmove(dst, pw, len(pw))
                p_response[0][i].resp = ctypes.cast(dst, ctypes.c_char_p)
                p_response[0][i].resp_retcode = 0
        return 0

    handle = ctypes.c_void_p()
    conv_struct = _PamConv(conv, None)
    try:
        rc = _libpam.pam_start(service.encode(), username.encode(),
                               ctypes.byref(conv_struct), ctypes.byref(handle))
        if rc != 0:
            return False
        rc = _libpam.pam_authenticate(handle, 0)
        if rc == 0:
            # also verify the account itself is valid (not expired/locked/disabled)
            rc = _libpam.pam_acct_mgmt(handle, 0)
        _libpam.pam_end(handle, rc)
        return rc == 0
    except Exception:
        return False


# ---------------- sessions ----------------
def make_token(user):
    payload = "%s|%d" % (user, int(time.time()) + SESSION_TTL)
    sig = hmac.new(SECRET, payload.encode(), hashlib.sha256).hexdigest()[:32]
    return base64.urlsafe_b64encode(("%s|%s" % (payload, sig)).encode()).decode()


def check_token(tok):
    try:
        raw = base64.urlsafe_b64decode(tok.encode()).decode()
        user, exp, sig = raw.rsplit("|", 2)
        payload = "%s|%s" % (user, exp)
        good = hmac.new(SECRET, payload.encode(), hashlib.sha256).hexdigest()[:32]
        if not hmac.compare_digest(sig, good):
            return None
        if int(exp) < time.time():
            return None
        return user
    except Exception:
        return None


# ---------------- telemetry / status ----------------
def read_all():
    out = {}
    try:
        txt = subprocess.run([HUD, "all"], capture_output=True, text=True, timeout=1.5).stdout
        for line in txt.splitlines():
            p = line.split(None, 1)
            if len(p) == 2:
                m = re.search(r"-?\d+(?:\.\d+)?", p[1])
                out[p[0]] = float(m.group()) if m else None
    except Exception:
        pass
    return out


def cpu_load(state):
    try:
        with open("/proc/stat") as fh:
            v = [int(x) for x in fh.readline().split()[1:]]
        idle = v[3] + (v[4] if len(v) > 4 else 0); total = sum(v)
        prev = state[0]; state[0] = (total, idle)
        if prev is None:
            return None
        dt = total - prev[0]; di = idle - prev[1]
        return round(max(0.0, min(100.0, 100.0 * (dt - di) / dt)), 1) if dt > 0 else None
    except Exception:
        return None



def interfacce_utente():
    u"""Le schede di rete che interessano a chi guarda, con tipo e indirizzo.

    ⚠️ Il tipo NON si indovina dal nome: «eth0» ed «enp3s0» sono convenzioni e
    una scheda si puo' rinominare. Si guarda /sys/class/net/<x>/wireless, che
    esiste solo se la scheda e' radio.
    """
    fuori = []
    try:
        base = "/sys/class/net"
        for nome in sorted(os.listdir(base)):
            if nome == "lo":
                continue
            # ⚠️ Si tengono solo le interfacce appoggiate a un dispositivo
            # vero: /sys/class/net/<x>/device esiste per PCI e USB e NON esiste
            # per ponti, veth, tun e tap. Prima si scartavano per prefisso del
            # nome — docker, veth, br-, virbr, tun, tap — e bastava un ponte
            # chiamato altrimenti per passare: `lxcbr0`, che arriva con Waydroid,
            # stava in fila con la scheda di rete come se fosse una di quelle.
            # Allungare l'elenco avrebbe solo rimandato il problema al prossimo
            # nome nuovo. E la funzione qui sopra dichiara gia' di non voler
            # indovinare dal nome: questo lo rispetta anche qui.
            #
            # ⚠️ ZeroTier e' l'eccezione, ed e' voluta: e' una tun e il metro la
            # scarterebbe, ma e' una rete che l'utente ha scelto e su cui lo
            # raggiungiamo. Il Wi-Fi USB invece passa da solo, perche' un
            # dispositivo ce l'ha.
            fisica = os.path.exists(os.path.join(base, nome, "device"))
            if not fisica and not nome.startswith("zt"):
                continue
            radio = os.path.isdir(os.path.join(base, nome, "wireless"))
            tipo = "wifi" if radio else ("zerotier" if nome.startswith("zt") else "cavo")
            try:
                with open(os.path.join(base, nome, "operstate")) as f:
                    su = f.read().strip() == "up"
            except OSError:
                su = False
            ind = []
            try:
                p = subprocess.run(["ip", "-o", "-4", "addr", "show", "dev", nome],
                                   capture_output=True, text=True, timeout=3)
                for riga in p.stdout.splitlines():
                    campi = riga.split()
                    if len(campi) >= 4:
                        ind.append(campi[3])
            except Exception:
                # Una riga di `ip` che non si lascia leggere: quell'indirizzo
                # non entra nell'elenco, l'interfaccia si', con quello che si e'
                # capito.
                pass
            fuori.append({"nome": nome, "tipo": tipo, "su": su, "indirizzi": ind})
    except OSError:
        # Nessuna rete da elencare: si risponde con l'elenco vuoto invece di far
        # cadere la pagina.
        pass
    return fuori



# ---------- GDDR6 memory temperature, and the secure processor ----------------
# The window in the app and this page read the same two things. Neither is
# reinvented here: the page shows what the daemon publishes and acts through the
# helper, exactly as the fan page does.
#
# ⚠️ THE READING IS A CAPPED SESSION, NOT A SENSOR. Polling the memory through
# the SMU wedges it - measured 19/09/2026: 31 min at one second, 55 at three, and
# only a reboot comes back from it. skillfish-gddr6-helper owns the ceiling and
# the cooldown; this file must never start the collector by itself.
GDDR6_HELPER = "/usr/local/bin/skillfish-gddr6-helper"
GDDR6_SNAPSHOT = "/run/bc250-memory/telemetry"
CCP_DRIVER = "/sys/bus/pci/drivers/ccp"
SICUREZZA_CHIAVI = ["fused_part", "boot_integrity", "debug_lock_on", "tsme_status",
                    "anti_rollback_status", "rpmc_production_enabled",
                    "rpmc_spirom_available", "hsp_tpm_available", "rom_armor_enforced"]


def gddr6_helper(azione, *extra):
    """Run the helper. The daemon is root, so no pkexec is involved."""
    try:
        p = subprocess.run([GDDR6_HELPER, azione] + [str(x) for x in extra],
                           capture_output=True, text=True, timeout=40)
        return json.loads((p.stdout or "{}").strip().splitlines()[-1])
    except Exception:
        return {"ok": False, "errore": "helper non disponibile"}


def gddr6_gradi():
    """(status, [temperature per chip]) from the collector's snapshot."""
    try:
        with open(GDDR6_SNAPSHOT, encoding="utf-8", errors="replace") as fh:
            righe = fh.read().splitlines()
    except OSError:
        return "assente", []
    if len(righe) < 4 or righe[0] != "BC250_MEMORY_V1":
        return "assente", []
    if righe[2] != "ok":
        return righe[2], []
    gradi = [(int(w) & 0xFF) * 2 - 40 for w in righe[3].split() if w.isdigit()]
    return ("ok", gradi) if len(gradi) == 8 else (righe[2], [])


def gddr6_stato():
    st = gddr6_helper("stato")
    stato, gradi = gddr6_gradi()
    fuori = {"attiva": bool(st.get("attiva")), "da": st.get("da"),
             "perche_no": st.get("perche_no"), "lettura": stato,
             "gradi": gradi}
    if gradi:
        fuori["media"] = round(sum(gradi) / float(len(gradi)), 1)
        fuori["caldo"] = max(gradi)
        fuori["caldo_chip"] = gradi.index(max(gradi))
    return fuori


def psp_sicurezza():
    """What the secure processor says about this board, or None where there is none.

    The device is found through the driver rather than by bus address: these files
    only exist with our ccp patches, and an address is not a promise.
    """
    try:
        indirizzi = [n for n in os.listdir(CCP_DRIVER) if n.count(":") == 2]
    except OSError:
        return None
    for indirizzo in sorted(indirizzi):
        base = os.path.join(CCP_DRIVER, indirizzo)
        fuori = {}
        for chiave in SICUREZZA_CHIAVI:
            try:
                with open(os.path.join(base, chiave)) as fh:
                    fuori[chiave] = fh.read().strip() == "1"
            except OSError:
                # Una chiave di sicurezza che questo firmware non espone: manca
                # dal quadro e le altre si mostrano lo stesso.
                pass
        if not fuori:
            continue
        try:
            with open(os.path.join(base, "bootloader_version")) as fh:
                fuori["bootloader"] = fh.read().strip() or None
        except OSError:
            fuori["bootloader"] = None
        return fuori
    return None


def sysinfo():
    def sh(c):
        try:
            return subprocess.run(c, shell=True, capture_output=True, text=True, timeout=2).stdout.strip()
        except Exception:
            return ""
    info = {}
    info["host"] = sh("hostname")
    # primary LAN IP = source address of the default route (not just the first of
    # `hostname -I`, which can be a secondary/stale or docker/lxc address).
    info["ip"] = (sh(r"ip -4 route get 1.1.1.1 2>/dev/null | sed -n 's/.*src \([0-9.]*\).*/\1/p'")
                  or (sh("hostname -I").split()[0] if sh("hostname -I") else ""))
    info["reti"] = interfacce_utente()
    info["gateway"] = sh(r"ip -4 route show default | head -1 | sed -n 's/.*via \([0-9.]*\).*/\1/p'")
    info["kernel"] = sh("uname -r")
    try:
        with open("/proc/uptime") as f:
            up = int(float(f.read().split()[0]))
        info["uptime"] = "%dd %dh %dm" % (up // 86400, (up % 86400) // 3600, (up % 3600) // 60)
    except Exception:
        info["uptime"] = ""
    try:
        with open("/proc/meminfo") as f:
            mi = {ln.split(":")[0]: int(ln.split()[1]) for ln in f if ":" in ln}
        info["ram_used_mb"] = (mi.get("MemTotal", 0) - mi.get("MemAvailable", 0)) // 1024
        info["ram_total_mb"] = mi.get("MemTotal", 0) // 1024
    except Exception:
        pass
    du = sh("df -h / | tail -1").split()
    if len(du) >= 5:
        info["disk_used"] = du[2]; info["disk_total"] = du[1]; info["disk_pct"] = du[4]
    info["cu"] = sh("cat /run/skillfish/cu_active 2>/dev/null")
    # ⚠️ Non solo quante, anche QUANDO l'ultima. Un conteggio da solo non si
    # puo' interpretare: quattro piantate di luglio, lette a dicembre, sembrano
    # un guasto di adesso. La riga del registro comincia con la data ISO.
    try:
        with open("/var/log/skillfish-freeze.log") as f:
            righe = [r for r in f.read().splitlines() if r.strip()]
        info["freezes"] = len(righe)
        info["freeze_last"] = righe[-1].split()[0] if righe else ""
    except Exception:
        info["freezes"] = 0
        info["freeze_last"] = ""
    return info


# ---------------- module backends ----------------
# ⚠️ The Control Center helper is a superset of the Tuner one (same protocol):
# it is preferred when installed, because only it knows the V/F governor.
TUNER_HELPER = ("/usr/local/bin/skillfish-cc-helper" if os.path.exists("/usr/local/bin/skillfish-cc-helper")
                else "/usr/local/bin/skillfish-tuner-helper")
PRESETS_F = "/usr/share/skillfish/tuner-presets.json"
REC_DIR_USER = None  # resolved at first use to the desktop user's home


def tuner_cmd(reqs, timeout=40):
    """Pipe one or more JSON command lines to the tuner helper; return last reply."""
    try:
        inp = "".join(json.dumps(r) + "\n" for r in reqs)
        p = subprocess.run([TUNER_HELPER], input=inp, capture_output=True, text=True, timeout=timeout)
        last = {"ok": False}
        for line in p.stdout.splitlines():
            try:
                last = json.loads(line)
            except Exception:
                pass
        return last
    except Exception as e:
        return {"ok": False, "error": str(e)}


def list_presets():
    try:
        with open(PRESETS_F) as f:
            return json.load(f).get("presets", [])
    except Exception:
        return []


def apply_preset(name):
    pr = next((p for p in list_presets() if p.get("name", "").lower() == (name or "").lower()), None)
    if not pr:
        return {"ok": False, "error": "preset sconosciuto"}
    c, g, fan = pr.get("cpu", {}), pr.get("gpu", {}), pr.get("fan", {})
    cmds = [
        {"cmd": "apply-cpu", "mhz": c.get("frequency"), "scale": c.get("scale"), "temp": c.get("max_temperature", 95)},
        {"cmd": "apply-gpu", "minmhz": g.get("min_mhz"), "minmv": g.get("min_mv"),
         "maxmhz": g.get("max_mhz"), "maxmv": g.get("max_mv")},
        {"cmd": "thermal-guard", "limit": pr.get("thermal_guard", 95)},
    ]
    # ⚠️ La ventola NON si scrive piu' qui dentro quando c'e' skillfish-fand:
    # il valore fisso del preset durerebbe un secondo e poi verrebbe riscritto
    # dal demone, e chi ha premuto il pulsante vedrebbe «applicato» senza che la
    # ventola faccia quello che il preset dice.
    if not ventola_demone():
        cmds.insert(2, {"cmd": "apply-fan", "mode": fan.get("mode", "auto"),
                        "pct": fan.get("pct", 45)})
    tuner_cmd(cmds)
    if ventola_demone():
        ventola_percentuale(fan.get("pct", 45), fan.get("mode", "auto"))
    return {"ok": True, "preset": pr.get("name")}


# ---- named Compute-Unit (WGP) profiles, shared by the web and native Tuner ----
CU_PROFILES_F = "/etc/skillfish/cu-profiles.json"


def cu_profiles_load():
    try:
        with open(CU_PROFILES_F) as f:
            d = json.load(f)
        return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def cu_profile_op(action, name, rows):
    name = (name or "").strip()[:40]
    profs = cu_profiles_load()
    if action == "save":
        if not name:
            return {"ok": False, "error": "nome mancante"}
        try:
            r = [int(x) & 0x1f for x in (rows or [])][:4]
        except Exception:
            r = []
        while len(r) < 4:
            r.append(0x1f)
        profs[name] = r
    elif action == "delete":
        profs.pop(name, None)
    elif action == "apply":
        r = profs.get(name)
        if not r:
            return {"ok": False, "error": "profilo sconosciuto"}
        res = tuner_cmd([{"cmd": "cu-apply", "rows": r}])
        res["profiles"] = profs
        return res
    else:
        return {"ok": False, "error": "azione sconosciuta"}
    try:
        os.makedirs(os.path.dirname(CU_PROFILES_F), exist_ok=True)
        with open(CU_PROFILES_F, "w") as f:
            json.dump(profs, f, indent=2)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    return {"ok": True, "profiles": profs}


# ---- temperature -> fan% curve (#3) ----
#
# ⚠️ DUE CONTROLLORI SULLO STESSO PWM: e' successo davvero, e si vedeva.
# Da qui girava un ciclo che scriveva la ventola ogni tre secondi; dal
# 21/08/2026 c'e' anche skillfish-fand, che la scrive ogni secondo con la sua
# curva, il suo anticipo e la sua emergenza. Misurato sulla scheda con le due
# curve diverse: il registro stava a 65 e ogni tre secondi saltava a 74 per un
# istante, poi tornava. Vinceva il demone perche' scrive piu' spesso, ma la
# ventola prendeva un calcio a ogni giro dell'altro.
#
# La regola adesso e' una sola: SE C'E' skillfish-fand, COMANDA LUI. Questo
# modulo diventa un telecomando — legge e scrive la SUA configurazione, e non
# tocca piu' nessun PWM. Cosi' dal web si ottiene lo stesso comportamento della
# finestra, anticipo ed emergenza compresi, invece di una seconda copia piu'
# debole.
#
# Il ciclo vecchio resta per le macchine dove skillfish-fand non c'e'
# (dashboard installata da sola, o versioni precedenti): li' e' meglio di
# niente. Si spegne da solo appena il demone compare, senza riavviare nulla.
FAN_CURVE_F = "/etc/skillfish/fan-curve.json"
# ---------- taratura automatica della curva GPU ----------
# Il lavoro vero lo fa skillfish-gpu-calibrate, fuori dal cruscotto: dura una
# mezz'ora, deve andare avanti anche se si chiude il browser, e deve poter
# essere lanciato anche da terminale su una macchina senza desktop.
#
# ⚠️ «In corso» NON si decide dal processo che abbiamo avviato noi. Se il
# cruscotto viene riavviato durante una taratura quel riferimento sparisce, e la
# pagina direbbe «ferma» mentre la GPU e' sotto carico da venti minuti. Si guarda
# invece il segnaposto che scrive il programma stesso, che sopravvive a tutto —
# ed e' lo stesso file su cui si regge il recupero all'avvio.
CALIBRA = "/usr/local/bin/skillfish-gpu-calibrate"
CALIBRA_LOG = "/var/lib/skillfish/gpu-calibrazione.log"
CALIBRA_SEGNO = "/var/lib/skillfish/gpu-calibrazione-in-corso"
CALIBRA_ESITO = "/var/lib/skillfish/gpu-calibrazione.json"


def calibra_stato(righe=80):
    d = {"ok": True, "in_corso": os.path.exists(CALIBRA_SEGNO), "righe": [], "esito": None}
    try:
        with open(CALIBRA_LOG, encoding="utf-8", errors="replace") as f:
            d["righe"] = f.read().splitlines()[-righe:]
    except Exception:
        pass
    try:
        with open(CALIBRA_ESITO, encoding="utf-8") as f:
            d["esito"] = json.load(f)
    except Exception:
        pass
    return d


def calibra_avvia():
    if os.path.exists(CALIBRA_SEGNO):
        return {"ok": False, "error": "una taratura e' gia' in corso"}
    if not os.path.exists(CALIBRA):
        return {"ok": False, "error": "manca %s" % CALIBRA}
    try:
        d = os.path.dirname(CALIBRA_LOG)
        if not os.path.isdir(d):
            os.makedirs(d)
        # ⚠️ Il file si apre in un with: Popen duplica il descrittore per il
        # figlio, e la copia del padre va chiusa. Senza, il cruscotto si
        # tiene un descrittore aperto a ogni taratura, per sempre.
        with open(CALIBRA_LOG, "w") as f:
            # start_new_session: la taratura non deve morire se muore il cruscotto
            subprocess.Popen([CALIBRA, "taratura"], stdout=f, stderr=subprocess.STDOUT,
                             stdin=subprocess.DEVNULL, start_new_session=True)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    return {"ok": True, "in_corso": True}


def calibra_ferma():
    # ⚠️ Fermare non basta: si resta con applicata la curva che si stava
    # PROVANDO, che e' quella che potrebbe non reggere. `ripristina` rimette
    # quella di partenza e toglie il segnaposto.
    try:
        subprocess.run(["pkill", "-f", "skillfish-gpu-calibrate taratura"], timeout=10)
        subprocess.run([CALIBRA, "ripristina"], timeout=60,
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    return calibra_stato()


SNAP_LEGGI = "/usr/local/bin/skillfish-snapshots-read"
SNAP_HELPER = "/usr/local/bin/skillfish-snapshots-helper"


def snapshots_leggi():
    u"""Gli snapshot del sistema che sta girando, letti dallo stesso programma
    che usa la finestra.

    ⚠️ Si riusa `skillfish-snapshots-read` invece di rileggere /.snapshots qui:
    il formato di info.xml, la data in UTC e l'ordine numerico sono gia' risolti
    la', e due letture diverse della stessa cosa finiscono sempre per divergere.

    ⚠️ Le righe BERSAGLIO/SNAPB (i sistemi installati visti da una sessione
    live) qui si IGNORANO: da remoto non si ripara un altro disco, vedi la nota
    in cima al modulo.
    """
    try:
        r = subprocess.run([SNAP_LEGGI], capture_output=True, text=True, timeout=60)
    except Exception as e:
        return {"ok": False, "errore": str(e), "snapshot": [], "daparte": []}
    if r.returncode != 0:
        return {"ok": False, "errore": (r.stderr or "").strip(),
                "snapshot": [], "daparte": []}

    snap, daparte, spazio = [], [], None
    for riga in (r.stdout or "").splitlines():
        p = riga.split("\t")
        p += [""] * (8 - len(p))
        if p[0] == "SNAP" and p[1]:
            snap.append({"num": p[1], "data": p[2], "tipo": p[3],
                         "pre": p[4], "descrizione": p[5]})
        elif p[0] == "SPAZIO" and p[1]:
            spazio = {"libero": p[1], "totale": p[2]}
        elif p[0] == "DAPARTE" and p[1]:
            daparte.append(p[1])
    # i piu' recenti in cima, come nella finestra
    try:
        snap.sort(key=lambda s: int(s["num"]), reverse=True)
    except Exception:
        pass
    return {"ok": True, "snapshot": snap, "daparte": daparte, "spazio": spazio}


def snapshots_azione(azione, numero=None, nota=None):
    u"""Crea, cancella, ripristina, annulla o fa pulizia.

    ⚠️ L'azione si controlla QUI contro un elenco chiuso, anche se l'aiutante la
    ricontrolla: questo e' il punto in cui arriva roba dalla rete, e la regola
    del progetto e' che un argomento da fuori si guarda prima di metterlo in un
    comando, non dopo.

    ⚠️ Niente --disco: da remoto si lavora solo sul sistema che sta girando.
    """
    if azione not in ("crea", "cancella", "ripristina", "annulla", "pulisci"):
        return {"ok": False, "errore": "unknown action"}
    cmd = [SNAP_HELPER, azione]
    if azione in ("cancella", "ripristina"):
        if not str(numero or "").isdigit():
            return {"ok": False, "errore": "that is not a snapshot number"}
        cmd.append(str(numero))
    elif azione == "crea":
        cmd.append((nota or "manual snapshot")[:120])
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=900)
    except Exception as e:
        return {"ok": False, "errore": str(e)}
    testo = ((r.stdout or "") + (r.stderr or "")).strip()
    return {"ok": r.returncode == 0, "uscita": testo}


VENTOLA_CONF = "/etc/skillfish/ventola.json"
VENTOLA_STATO = "/run/skillfish/ventola.json"
VENTOLA_HELPER = "/usr/local/bin/skillfish-fan-helper"
# On by default: left to itself the board's EC parks the fan at a flat ~50% duty
# whatever the temperature, which caps sustained CPU/GPU clocks. Measured on the
# board, 100% duty holds the CPU 4 °C cooler than 31% under the same load, so the
# curve reaches 100% before the 85 °C thermal cap rather than at 90 °C.
FAN_CURVE_DEFAULT = {"enabled": True, "source": "max",
                     "points": [[40, 25], [55, 40], [68, 60], [76, 85], [83, 100]]}
_FAN = {"last_pct": None, "was_enabled": False}


def ventola_demone():
    u"""C'e' skillfish-fand, e sta girando? Allora comanda lui."""
    try:
        return subprocess.call(["systemctl", "is-active", "--quiet",
                                "skillfish-fand.service"]) == 0
    except Exception:
        return False


def ventola_stato():
    u"""Quello che il demone pubblica: letture, storia, motivo."""
    try:
        with open(VENTOLA_STATO) as f:
            return json.load(f)
    except Exception:
        return {}


def ventola_conf():
    try:
        with open(VENTOLA_CONF) as f:
            return json.load(f)
    except Exception:
        return {}


def ventola_scrivi(conf):
    u"""La configurazione passa SEMPRE dall'helper, mai scritta a mano.

    L'helper rifa' ogni campo da zero e fa rispettare i limiti — il minimo che
    non scende sotto la soglia, la curva che non puo' scendere mentre la
    temperatura sale. Scrivere il file da qui vorrebbe dire avere due posti in
    cui si controllano gli stessi numeri, e prima o poi uno dei due si scorda
    un controllo."""
    if not os.path.exists(VENTOLA_HELPER):
        return {"ok": False, "error": "manca skillfish-fan-helper"}
    try:
        p = subprocess.run([VENTOLA_HELPER, "scrivi"], input=json.dumps(conf),
                           capture_output=True, text=True, timeout=60)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    if p.returncode != 0:
        return {"ok": False, "error": (p.stderr or p.stdout or "").strip()}
    return {"ok": True}


VENTOLA_PROVA = "/etc/skillfish/ventola-prova.json"


def ventola_prova_esito():
    u"""L'esito dell'ultima prova del PWM, se e' mai stata fatta.

    Su molte schede — server e portatili soprattutto — scrivere il PWM viene
    accettato e non muove niente. E' l'informazione che distingue un pannello
    che funziona da un pannello che sembra funzionare, quindi il web la mostra
    come la mostra la finestra."""
    try:
        with open(VENTOLA_PROVA) as f:
            return json.load(f)
    except Exception:
        return {}


def ventola_etichette(d):
    u"""I nomi che l'utente ha dato ai sensori, scritti dall'helper."""
    if not os.path.exists(VENTOLA_HELPER):
        return {"ok": False, "error": "manca skillfish-fan-helper"}
    try:
        p = subprocess.run([VENTOLA_HELPER, "etichette"], input=json.dumps(d),
                           capture_output=True, text=True, timeout=30)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    return ({"ok": True} if p.returncode == 0
            else {"ok": False, "error": (p.stderr or p.stdout or "").strip()})


def ventola_prova(pwm, fan):
    u"""La prova che dice se quel PWM muove davvero quella ventola.

    ⚠️ Dura una trentina di secondi e ferma il controllo mentre misura. Da qui
    si lancia con un tempo massimo generoso: se scadesse a meta', l'helper
    rimette comunque tutto com'era nel suo `finally`, ma il web resterebbe senza
    risposta e l'utente non saprebbe com'e' finita."""
    if not os.path.exists(VENTOLA_HELPER):
        return {"ok": False, "error": "manca skillfish-fan-helper"}
    try:
        p = subprocess.run([VENTOLA_HELPER, "prova"],
                           input=json.dumps({"pwm": pwm, "fan": fan}),
                           capture_output=True, text=True, timeout=180)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    if p.returncode != 0:
        return {"ok": False, "error": (p.stderr or p.stdout or "").strip()}
    try:
        return {"ok": True, "esito": json.loads(p.stdout)}
    except Exception:
        return {"ok": False, "error": p.stdout[:200]}


def fan_curve_load():
    if ventola_demone():
        c = ventola_conf()
        return {"enabled": bool(c.get("attivo")),
                # il demone prende SEMPRE la piu' alta fra le sorgenti scelte,
                # che e' esattamente cio' che qui si chiamava "max"
                "source": "max",
                "points": [[int(round(p[0])), int(round(p[1]))]
                           for p in (c.get("curva") or FAN_CURVE_DEFAULT["points"])]}
    try:
        with open(FAN_CURVE_F) as f:
            d = json.load(f)
        out = dict(FAN_CURVE_DEFAULT)
        out.update({k: d[k] for k in ("enabled", "source", "points") if k in d})
        return out
    except Exception:
        return dict(FAN_CURVE_DEFAULT)


def fan_curve_save(d):
    if ventola_demone():
        c = ventola_conf()
        if "enabled" in d:
            c["attivo"] = bool(d["enabled"])
        if isinstance(d.get("points"), list) and d["points"]:
            c["curva"] = [[float(p[0]), float(p[1])] for p in d["points"]]
            c["preset"] = "personalizzata"
        # ⚠️ `source` non si traduce e viene ignorato apposta. Qui erano tre
        # scelte fisse (cpu/gpu/max); il demone lascia scegliere QUALUNQUE
        # canale vero, e sono elencati per nome nella sua configurazione.
        # Sovrascriverli con "cpu" o "gpu" cancellerebbe una scelta piu' fine
        # con una piu' grossolana.
        r = ventola_scrivi(c)
        if not r.get("ok"):
            return r
        return {"ok": True, "curve": fan_curve_load(), "preview": fan_curve_pct()}
    cur = fan_curve_load()
    if "enabled" in d:
        cur["enabled"] = bool(d["enabled"])
    if d.get("source") in ("max", "gpu", "cpu"):
        cur["source"] = d["source"]
    if isinstance(d.get("points"), list):
        pts = []
        for p in d["points"]:
            try:
                pts.append([max(0, min(110, int(p[0]))), max(0, min(100, int(p[1])))])
            except Exception:
                pass
        if pts:
            cur["points"] = sorted(pts)
    try:
        os.makedirs(os.path.dirname(FAN_CURVE_F), exist_ok=True)
        with open(FAN_CURVE_F, "w") as f:
            json.dump(cur, f, indent=2)
    except Exception as e:
        return {"ok": False, "error": str(e)}
    return {"ok": True, "curve": cur, "preview": fan_curve_pct(cur)}


def fan_curve_pct(cur=None):
    """Interpolate the target fan % for the current temperature."""
    if cur is None and ventola_demone():
        return ventola_stato().get("duty")
    cur = cur or fan_curve_load()
    v = read_all()
    g = v.get("gpu_temp") if isinstance(v.get("gpu_temp"), (int, float)) else 0
    c = v.get("cpu_temp") if isinstance(v.get("cpu_temp"), (int, float)) else 0
    t = {"gpu": g, "cpu": c}.get(cur["source"], max(g, c))
    pts = sorted(cur["points"])
    if not pts:
        return None
    if t <= pts[0][0]:
        return pts[0][1]
    if t >= pts[-1][0]:
        return pts[-1][1]
    for i in range(1, len(pts)):
        t0, p0 = pts[i - 1]
        t1, p1 = pts[i]
        if t0 <= t <= t1:
            return round(p0 + (p1 - p0) * (t - t0) / (t1 - t0)) if t1 != t0 else p1
    return pts[-1][1]


def ventola_percentuale(pct, modo="manual"):
    u"""Una percentuale fissa, detta al demone invece che scritta sul registro.

    ⚠️ Con skillfish-fand attivo, scrivere il PWM da qui non serve a niente: al
    giro dopo lo riscrive lui. Una percentuale fissa si ottiene con una curva
    piatta — due punti alla stessa altezza — che e' esattamente la stessa cosa
    detta nella lingua del demone. «auto» invece vuol dire restituire la ventola
    al firmware, cioe' spegnere il controllo.
    """
    c = ventola_conf()
    if modo == "auto":
        c["attivo"] = False
    else:
        p = max(20, min(100, int(pct)))
        c["attivo"] = True
        c["curva"] = [[0.0, float(p)], [100.0, float(p)]]
        c["preset"] = "personalizzata"
    r = ventola_scrivi(c)
    return {"ok": bool(r.get("ok")), "error": r.get("error", ""),
            "pct": None if modo == "auto" else int(pct), "mode": modo}


def fan_curve_loop():
    while True:
        try:
            # ⚠️ Il controllo vero e' di skillfish-fand quando c'e'. Qui non si
            # scrive piu' niente: due programmi sullo stesso registro fanno
            # oscillare la ventola, e l'abbiamo visto succedere.
            if ventola_demone():
                if _FAN["was_enabled"]:
                    _FAN["was_enabled"] = False
                    _FAN["last_pct"] = None
                time.sleep(5)
                continue
            cur = fan_curve_load()
            if cur["enabled"]:
                pct = fan_curve_pct(cur)
                if pct is not None and (_FAN["last_pct"] is None or abs(pct - _FAN["last_pct"]) >= 3):
                    tuner_cmd([{"cmd": "apply-fan", "mode": "manual", "pct": int(pct)}])
                    _FAN["last_pct"] = pct
                _FAN["was_enabled"] = True
            elif _FAN["was_enabled"]:
                tuner_cmd([{"cmd": "apply-fan", "mode": "auto"}])  # hand the fan back to firmware
                _FAN["was_enabled"] = False
                _FAN["last_pct"] = None
        except Exception:
            pass
        time.sleep(3)



_TOPO = {}


def cpu_threads():
    """[[cpu, core, MHz|None], ...] for the telemetry stream.

    MHz is None when the thread is parked (Tuner core toggle): an offline CPU
    disappears from /proc/cpuinfo and loses its topology/ dir, so the physical
    core comes from the map cached while it was still up."""
    base = "/sys/devices/system/cpu"
    mhz = {}
    try:
        cur = None
        with open("/proc/cpuinfo") as f:
            for ln in f:
                if ln.startswith("processor"):
                    cur = int(ln.split(":")[1])
                elif ln.lower().startswith("cpu mhz") and cur is not None:
                    mhz[cur] = round(float(ln.split(":")[1]))
    except Exception:
        return []
    try:
        cpus = sorted(int(d[3:]) for d in os.listdir(base) if re.match(r"cpu\d+$", d))
    except Exception:
        return []
    out = []
    for c in cpus:
        try:
            with open("%s/cpu%d/online" % (base, c)) as f:
                on = f.read().strip() == "1"
        except Exception:
            on = True                      # cpu0 exposes no 'online' node
        try:
            with open("%s/cpu%d/topology/core_id" % (base, c)) as f:
                _TOPO[c] = int(f.read().strip())
        except Exception:
            pass
        out.append([c, _TOPO.get(c, c // 2), mhz.get(c) if on else None])
    return out


def cpu_coremap():
    """Physical CPU core layout + live per-core MHz (for affinity / pinning)."""
    base = "/sys/devices/system/cpu"

    def rd(p):
        try:
            with open(p) as f:
                return f.read().strip()
        except Exception:
            return ""
    # live MHz per logical CPU from /proc/cpuinfo
    mhz = {}
    try:
        cur = None
        with open("/proc/cpuinfo") as f:
            for ln in f:
                if ln.startswith("processor"):
                    cur = int(ln.split(":")[1])
                elif ln.lower().startswith("cpu mhz") and cur is not None:
                    mhz[cur] = round(float(ln.split(":")[1]))
    except Exception:
        pass
    out = []
    try:
        cpus = sorted(int(d[3:]) for d in os.listdir(base) if re.match(r"cpu\d+$", d))
    except Exception:
        cpus = []
    for c in cpus:
        t = "%s/cpu%d/topology" % (base, c)
        out.append({"cpu": c, "core": rd(t + "/core_id"),
                    "pkg": rd(t + "/physical_package_id"),
                    "siblings": rd(t + "/thread_siblings_list"),
                    "mhz": mhz.get(c, "")})
    return {"ok": True, "cpus": out, "n": len(out)}


_DESKTOP_USER = {"t": 0.0, "u": ""}


def desktop_user():
    """The account the desktop actions run as (issue #110).

    ⚠️ /etc/skillfish/dashboard.json ships "user": "skillfish", and nothing
    replaces it with the account created at install time. On a board where that
    account does not exist, every `sudo -u skillfish` failed at once: the KVM
    page showed "Failed to connect", the launcher did nothing.

    So: the configured user if it exists; else the owner of the graphical
    session on a seat (the SDDM greeter has class "greeter" and is skipped);
    else the first real account. Cached for 30 s, since this is asked on every
    launch and loginctl is a process each time.
    """
    import pwd
    now = time.monotonic()
    if _DESKTOP_USER["u"] and now - _DESKTOP_USER["t"] < 30:
        return _DESKTOP_USER["u"]
    scelto = ""
    u = CONFIG.get("user") or ""
    try:
        pwd.getpwnam(u)
        scelto = u
    except KeyError:
        pass
    if not scelto:
        try:
            ids = subprocess.run(["loginctl", "list-sessions", "--no-legend"],
                                 capture_output=True, text=True, timeout=5).stdout
            for ln in ids.splitlines():
                f = ln.split()
                if not f:
                    continue
                p = dict(kv.split("=", 1) for kv in subprocess.run(
                    ["loginctl", "show-session", f[0], "-p", "Name", "-p", "Class",
                     "-p", "Type", "-p", "Seat"],
                    capture_output=True, text=True, timeout=5).stdout.splitlines() if "=" in kv)
                if (p.get("Class") == "user" and p.get("Seat")
                        and p.get("Type") in ("x11", "wayland")
                        and p.get("Name") not in ("", "root")):
                    scelto = p["Name"]
                    break
        except (OSError, subprocess.SubprocessError, ValueError):
            # no logind to ask: fall through to the first real account
            pass
    if not scelto:
        reali = [p.pw_name for p in pwd.getpwall()
                 if 1000 <= p.pw_uid < 65534 and os.path.isdir(p.pw_dir)]
        scelto = reali[0] if reali else u or "skillfish"
    _DESKTOP_USER.update(t=now, u=scelto)
    return scelto


def user_home():
    global REC_DIR_USER
    if REC_DIR_USER:
        return REC_DIR_USER
    u = desktop_user()
    try:
        import pwd
        REC_DIR_USER = pwd.getpwnam(u).pw_dir
    except Exception:
        REC_DIR_USER = "/home/" + u
    return REC_DIR_USER


def user_env():
    """Env to launch GUI apps on the desktop user's session."""
    return {"DISPLAY": ":0", "XDG_RUNTIME_DIR": "/run/user/1000",
            "DBUS_SESSION_BUS_ADDRESS": "unix:path=/run/user/1000/bus",
            "PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": user_home()}



# ======================= Control Center: games, profiles, governor =======================
# The window does these things on the machine; here they are done for the
# desktop user from the web. Overrides are FILES in that user's home, so they
# are written as root and handed back to the user.
CC_MESA_ICD = "/opt/skillfish-gfx1013/share/vulkan/icd.d/radeon_icd.x86_64.json"
CC_MESA_LIBS = "/opt/skillfish-gfx1013/lib/x86_64-linux-gnu"
CC_MESA_NOSTRE = ("VK_DRIVER_FILES", "LD_LIBRARY_PATH")
CC_APP = {"steam": "com.valvesoftware.Steam", "heroic": "com.heroicgameslauncher.hgl"}
CC_FSR4 = ("PROTON_USE_OPTISCALER", "PROTON_FSR4_UPGRADE")
CC_PROFILI_SISTEMA = "/usr/share/skillfish/profili.json"
CC_CURVA_PREDEFINITA = "/usr/share/skillfish/vf-curva-predefinita.json"
CC_PRESET_VENTOLA = {
    "silenzioso": [[45, 20], [60, 30], [70, 45], [80, 70], [88, 100]],
    "equilibrato": [[40, 30], [55, 40], [65, 60], [75, 85], [85, 100]],
    "prestazioni": [[35, 40], [50, 60], [60, 80], [70, 100], [85, 100]],
    "massimo": [[0, 100], [100, 100]],
}


def _cc_utente():
    import pwd
    u = desktop_user()
    try:
        p = pwd.getpwnam(u)
        return p.pw_uid, p.pw_gid, p.pw_dir
    except Exception:
        return None, None, "/home/" + u


def _cc_override(app):
    import configparser
    c = configparser.RawConfigParser()
    c.optionxform = str
    c.read(os.path.join(user_home(), ".local/share/flatpak/overrides", CC_APP[app]))
    for s in ("Context", "Environment"):
        if not c.has_section(s):
            c.add_section(s)
    return c


def _cc_override_scrivi(app, c):
    uid, gid, casa = _cc_utente()
    cartella = os.path.join(casa, ".local/share/flatpak/overrides")
    os.makedirs(cartella, exist_ok=True)
    percorso = os.path.join(cartella, CC_APP[app])
    with open(percorso, "w", encoding="utf-8", newline="\n") as fh:
        c.write(fh, space_around_delimiters=False)
    if uid is not None:
        try:
            os.chown(percorso, uid, gid)
            os.chown(cartella, uid, gid)
        except OSError:
            # not running as root, or the user is gone: the file is still written
            pass


def _cc_pulisci_unset(c, nomi):
    v = c.get("Context", "unset-environment", fallback="")
    resto = [x for x in v.split(";") if x and x not in nomi]
    if resto:
        c.set("Context", "unset-environment", ";".join(resto) + ";")
    elif c.has_option("Context", "unset-environment"):
        c.remove_option("Context", "unset-environment")


def _cc_override_sistema(app):
    """The system-wide override, which is where the package writes from 26.09.4."""
    import configparser
    c = configparser.RawConfigParser()
    c.optionxform = str
    c.read(os.path.join("/var/lib/flatpak/overrides", CC_APP[app]))
    for s in ("Context", "Environment"):
        if not c.has_section(s):
            c.add_section(s)
    return c


def cc_mesa_stato(app):
    # ⚠️ TWO FILES, IN THIS ORDER. flatpak lets a user override beat the system
    # one, so the user's file decides when it sets the variable at all; when it
    # does not, the answer is in /var/lib/flatpak/overrides, and reading only
    # the first one made the page say "stock" while the games ran on ours.
    for c in (_cc_override(app), _cc_override_sistema(app)):
        icd = c.get("Environment", "VK_DRIVER_FILES", fallback=None)
        if icd and "VK_DRIVER_FILES" not in c.get("Context", "unset-environment", fallback=""):
            return icd == CC_MESA_ICD
    return False


def cc_mesa_set(app, on):
    if app not in CC_APP:
        return {"ok": False, "err": "app sconosciuta"}
    if on and not os.path.exists(CC_MESA_ICD):
        return {"ok": False, "err": "la nostra Mesa non e' installata"}
    c = _cc_override(app)
    _cc_pulisci_unset(c, CC_MESA_NOSTRE)
    if on:
        v = c.get("Context", "filesystems", fallback="")
        voci = [x for x in v.split(";") if x]
        if "/opt/skillfish-gfx1013:ro" not in voci:
            voci.append("/opt/skillfish-gfx1013:ro")
        c.set("Context", "filesystems", ";".join(voci) + ";")
        c.set("Environment", "VK_DRIVER_FILES", CC_MESA_ICD)
        c.set("Environment", "LD_LIBRARY_PATH", CC_MESA_LIBS)
    else:
        for n in CC_MESA_NOSTRE:
            if c.has_option("Environment", n):
                c.remove_option("Environment", n)
    _cc_override_scrivi(app, c)
    return {"ok": True, "on": bool(on)}


def cc_fsr4_stato(app):
    c = _cc_override(app)
    return all(c.get("Environment", k, fallback="") == "1" for k in CC_FSR4)


def cc_fsr4_set(app, on):
    if app not in CC_APP:
        return {"ok": False, "err": "app sconosciuta"}
    c = _cc_override(app)
    _cc_pulisci_unset(c, CC_FSR4)
    for k in CC_FSR4:
        if on:
            c.set("Environment", k, "1")
        elif c.has_option("Environment", k):
            c.remove_option("Environment", k)
    _cc_override_scrivi(app, c)
    return {"ok": True, "on": bool(on)}


def _cc_proton_dir(dove):
    casa = user_home()
    return (os.path.join(casa, ".var/app/com.valvesoftware.Steam/data/Steam/compatibilitytools.d") if dove == "steam"
            else os.path.join(casa, ".var/app/com.heroicgameslauncher.hgl/config/heroic/tools/proton"))


def _cc_proton_nome_interno(dove, n):
    """The name Steam knows the tool by, read from compatibilitytool.vdf.

    ⚠️ Not the folder name: GE-Proton11-6 declares itself GE-Proton11-6-x86_64,
    and a name Steam does not know means the game starts with no Proton at all.
    """
    import re as _re
    # il nome arriva anche dalla rete: qui si pretende che sia un nome di
    # cartella e nient'altro, cosi' nessun ".." puo' uscire da tools.d
    # il nome arriva anche dalla rete. Non basta guardarne la forma: si accetta
    # solo se e' una delle cartelle che esistono davvero li' dentro, cosi' non
    # c'e' percorso che possa uscire da tools.d
    base = os.path.realpath(_cc_proton_dir(dove))
    try:
        if n not in os.listdir(base):
            return n
    except OSError:
        return n
    # e comunque si guarda dove si e' finiti: il percorso deve stare dentro la
    # cartella degli strumenti, altrimenti non lo si apre
    dentro = os.path.realpath(os.path.join(base, n, "compatibilitytool.vdf"))
    if not dentro.startswith(base + os.sep):
        return n
    try:
        with open(dentro, encoding="utf-8", errors="replace") as f:
            m = _re.search(r'"compat_tools"\s*\{\s*"([^"]+)"', f.read())
        if m:
            return m.group(1)
    except OSError:
        # no compatibilitytool.vdf, or unreadable: fall back to the folder name
        pass
    return n


def cc_proton():
    import re as _re
    out = {}
    for dove in ("steam", "heroic"):
        try:
            out[dove] = sorted(d for d in os.listdir(_cc_proton_dir(dove)) if os.path.isdir(os.path.join(_cc_proton_dir(dove), d)))
        except OSError:
            out[dove] = []
    out["default_heroic"] = ""
    try:
        with open(os.path.join(user_home(), ".var/app/com.heroicgameslauncher.hgl/config/heroic/config.json")) as f:
            out["default_heroic"] = (json.load(f).get("defaultSettings", {}).get("wineVersion") or {}).get("name", "")
    except (OSError, json.JSONDecodeError):
        # Heroic not installed, or its config is mid-write: report the "" above
        out["default_heroic"] = ""
    out["default_steam"] = ""
    out["nomi_interni"] = {d: _cc_proton_nome_interno("steam", d) for d in out.get("steam", [])}
    try:
        with open(os.path.join(user_home(), ".var/app/com.valvesoftware.Steam/data/Steam/config/config.vdf"), encoding="utf-8", errors="replace") as f:
            m = _re.search(r'"CompatToolMapping"\s*\{(.*?)\n\t{4}\}', f.read(), _re.S)
        if m:
            z = _re.search(r'"0"\s*\{[^}]*?"name"\s*"([^"]*)"', m.group(1), _re.S)
            if z:
                out["default_steam"] = z.group(1)
    except OSError:
        # Steam not installed, or no config.vdf yet: report the "" above
        out["default_steam"] = ""
    return out


def cc_proton_default(dove, nome):
    import re as _re
    import shutil as _sh
    nome = (nome or "").strip()
    if dove not in ("steam", "heroic") or not _re.match(r"^[A-Za-z0-9._-]+$", nome):
        return {"ok": False, "err": "richiesta non valida"}
    if nome not in cc_proton().get(dove, []):
        return {"ok": False, "err": "non installata in %s" % dove}
    uid, gid, casa = _cc_utente()
    if dove == "heroic":
        p = os.path.join(casa, ".var/app/com.heroicgameslauncher.hgl/config/heroic/config.json")
        try:
            with open(p) as f:
                d = json.load(f)
            d.setdefault("defaultSettings", {})["wineVersion"] = {
                "bin": os.path.join(_cc_proton_dir("heroic"), nome, "proton"), "name": nome, "type": "proton"}
            with open(p, "w") as f:
                json.dump(d, f, indent=2)
            return {"ok": True}
        except Exception as e:
            return {"ok": False, "err": str(e)}
    # from here down it is Steam, which wants the name declared inside the tool
    nome = _cc_proton_nome_interno("steam", nome)
    try:
        aperto = "com.valvesoftware.Steam" in subprocess.run(["flatpak", "ps", "--columns=application"], capture_output=True, text=True, timeout=10).stdout
    except Exception:
        aperto = False
    if aperto:
        return {"ok": False, "err": "chiudi Steam prima: riscrive la configurazione all'uscita"}
    p = os.path.join(casa, ".var/app/com.valvesoftware.Steam/data/Steam/config/config.vdf")
    try:
        with open(p, encoding="utf-8", errors="replace") as f:
            t = f.read()
    except OSError:
        return {"ok": False, "err": "config.vdf non trovato"}
    voce = '"0"\n\t\t\t\t\t{\n\t\t\t\t\t\t"name"\t\t"%s"\n\t\t\t\t\t\t"config"\t\t""\n\t\t\t\t\t\t"priority"\t\t"75"\n\t\t\t\t\t}' % nome
    m = _re.search(r'"CompatToolMapping"\s*\{', t)
    if m:
        z = _re.search(r'"0"\s*\{[^}]*\}', t[m.end():], _re.S)
        if z and z.start() < 400:
            t = t[:m.end() + z.start()] + voce + t[m.end() + z.end():]
        else:
            t = t[:m.end()] + "\n\t\t\t\t\t" + voce + t[m.end():]
    else:
        s = _re.search(r'"Steam"\s*\{', t)
        if not s:
            return {"ok": False, "err": "config.vdf senza blocco Steam"}
        t = t[:s.end()] + '\n\t\t\t\t"CompatToolMapping"\n\t\t\t\t{\n\t\t\t\t\t' + voce + "\n\t\t\t\t}" + t[s.end():]
    _sh.copy2(p, p + ".skillfish.bak")
    with open(p, "w", encoding="utf-8") as f:
        f.write(t)
    return {"ok": True}


def cc_giochi():
    return {"ok": True,
            "mesa": {a: cc_mesa_stato(a) for a in CC_APP},
            "mesa_sistema": tuner_cmd([{"cmd": "mesa-sistema"}]),
            "scx": tuner_cmd([{"cmd": "scx"}]),
            "fsr4": {a: cc_fsr4_stato(a) for a in CC_APP},
            "proton": cc_proton()}


# ---- the CPU test with countdown and Stop (web side) ----------------------------
# The helper's own test-cpu/suggest-uv block for a minute or two and the page
# could only wait. Here the steps are run in a thread: the helper applies the
# values WITHOUT writing them to disk (cpu-volatile), sysbench loads every
# thread, the lowest clock is watched, and the page polls the state once a
# second. Stop kills the load; whatever happens the applied values come back.
def _cpu_min_mhz():
    try:
        with open("/proc/cpuinfo") as fh:
            fs = [float(ln.split(":")[1]) for ln in fh if ln.lower().startswith("cpu mhz")]
        return min(fs) if fs else None
    except Exception:
        return None


class ProvaCpuWeb:
    TOLLERANZA = 200

    def __init__(self):
        self.lock = threading.Lock()
        self.attiva = False
        self.stop = False
        self.etichetta = ""
        self.resto = 0
        self.esiti = []
        self.migliore = None
        self.finita = False
        self.fermata = False
        self.proc = None
        self.temp = 85

    def stato(self):
        with self.lock:
            return {"ok": True, "attiva": self.attiva, "etichetta": self.etichetta, "resto": self.resto,
                    "esiti": list(self.esiti), "migliore": self.migliore, "finita": self.finita,
                    "fermata": self.fermata}

    def avvia(self, passi, temp):
        with self.lock:
            if self.attiva:
                return {"ok": False, "err": "prova in corso"}
            self.attiva, self.stop, self.finita, self.fermata = True, False, False, False
            self.esiti, self.migliore, self.etichetta, self.resto = [], None, "", 0
            self.temp = int(temp)
        th = threading.Thread(target=self._giro, args=(list(passi),), daemon=True)
        th.start()
        return {"ok": True}

    def ferma(self):
        with self.lock:
            self.stop = True
            p = self.proc
        if p is not None:
            try:
                p.terminate()
            except OSError:
                # already exited: nothing left to stop
                pass
        return {"ok": True}

    def _giro(self, passi):
        nth = os.cpu_count() or 8
        for mhz, scale, secs in passi:
            if self.stop:
                break
            with self.lock:
                self.etichetta, self.resto = "%d MHz  ·  UV %d" % (mhz, scale), secs
            r = tuner_cmd([{"cmd": "cpu-volatile", "mhz": int(mhz), "scale": int(scale), "temp": self.temp}])
            if not r.get("ok"):
                with self.lock:
                    self.esiti.append({"mhz": mhz, "scale": scale, "tenuto": False, "min": 0, "err": r.get("err", "apply")})
                break
            try:
                proc = subprocess.Popen(["sysbench", "cpu", "--threads=%d" % nth, "--time=%d" % int(secs),
                                         "--cpu-max-prime=20000", "run"],
                                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            except OSError as e:
                with self.lock:
                    self.esiti.append({"mhz": mhz, "scale": scale, "tenuto": False, "min": 0, "err": str(e)})
                break
            with self.lock:
                self.proc = proc
            minimo, t0 = None, time.monotonic()
            while proc.poll() is None and not self.stop:
                passati = time.monotonic() - t0
                if passati > 5:
                    m = _cpu_min_mhz()
                    if m is not None and (minimo is None or m < minimo):
                        minimo = m
                with self.lock:
                    self.resto = max(0, int(secs - passati))
                time.sleep(0.5)
            rc = proc.wait() if proc.poll() is not None else -1
            with self.lock:
                self.proc = None
            if self.stop:
                break
            tenuto = rc == 0 and minimo is not None and minimo >= mhz - self.TOLLERANZA
            with self.lock:
                self.esiti.append({"mhz": mhz, "scale": scale, "tenuto": tenuto, "min": int(minimo or 0)})
                if tenuto:
                    self.migliore = [mhz, scale]
            if not tenuto:
                break
        # back to the applied values, whatever happened
        try:
            cfg = (tuner_cmd([{"cmd": "get"}]).get("data") or {}).get("cpu") or {}
            tuner_cmd([{"cmd": "apply-cpu", "mhz": int(cfg.get("frequency", 3500)),
                        "scale": int(cfg.get("scale", 0)), "temp": int(cfg.get("max_temperature", 95))}])
        except Exception:
            # best-effort restore after the trial: a failure here should not crash the run
            pass
        with self.lock:
            self.fermata = self.stop
            self.attiva, self.finita, self.etichetta, self.resto = False, True, "", 0


PROVA_CPU = ProvaCpuWeb()


def cc_prova_cpu_post(data):
    az = data.get("azione", "avvia")
    if az == "stop":
        return PROVA_CPU.ferma()
    try:
        temp = int(data.get("temp", 85))
        if az == "uv":
            mhz = int(data.get("mhz", 3500))
            passi = [(mhz, s, 12) for s in range(0, -41, -2)]
        elif az == "max":
            passi = [(3600, -8, 30), (3700, -16, 30), (3800, -20, 30), (3900, -24, 30), (4000, -36, 30)]
        else:
            passi = [(int(data.get("mhz", 3500)), int(data.get("scale", 0)), 60)]
    except (TypeError, ValueError):
        return {"ok": False, "err": "valori non validi"}
    return PROVA_CPU.avvia(passi, temp)


def cc_coreunlock():
    r = tuner_cmd([{"cmd": "core-unlock"}])
    e = tuner_cmd([{"cmd": "coreunlock-efi"}])
    return {"ok": True, "abilitato": bool(r.get("abilitato")), "supportato": bool(r.get("supportato")),
            "efi": e if e.get("ok") else {"supportato": False}}


def cc_coreunlock_post(data):
    if "efi" in data:
        return tuner_cmd([{"cmd": "coreunlock-efi-set", "on": bool(data.get("efi"))}])
    return tuner_cmd([{"cmd": "core-unlock-set", "on": bool(data.get("on"))}])


def cc_gov():
    r = tuner_cmd([{"cmd": "gov-get"}])
    try:
        for h in os.listdir("/sys/class/hwmon"):
            p = "/sys/class/hwmon/" + h
            with open(p + "/name") as f:
                nome_chip = f.read().strip()
            if nome_chip == "amdgpu":
                with open(p + "/in0_input") as f:
                    r["mv"] = int(f.read().strip())
                break
    except (OSError, ValueError):
        # no amdgpu hwmon node, or it went away mid-read: report without "mv"
        pass
    try:
        with open(CC_CURVA_PREDEFINITA) as f:
            r["predefinita"] = json.load(f)
    except Exception:
        r["predefinita"] = {}
    return r


def cc_gov_post(data):
    az = data.get("azione")
    if az == "attiva":
        return tuner_cmd([{"cmd": "gov-attiva", "on": bool(data.get("on", True))}], timeout=90)
    if az in ("set", "prova"):
        return tuner_cmd([{"cmd": "gov-" + az, "conf": data.get("conf") or {}}], timeout=90)
    if az in ("conferma", "annulla"):
        return tuner_cmd([{"cmd": "gov-" + az}], timeout=90)
    return {"ok": False, "err": "azione sconosciuta"}


def _cc_conf_governor():
    try:
        with open("/etc/skillfish-vf-governor.json") as f:
            return json.load(f)
    except Exception:
        return {}


def cc_profili():
    import re as _re
    profili = {}
    for p in (CC_PROFILI_SISTEMA, os.path.join(user_home(), ".config/skillfish/profili.json")):
        try:
            with open(p) as f:
                d = json.load(f)
            if isinstance(d, dict):
                profili.update(d)
        except (OSError, json.JSONDecodeError):
            # this profiles file is missing or malformed: skip it, keep the rest
            pass
    g = _cc_conf_governor()
    fan = ventola_conf() or {}
    m = None
    try:
        with open("/etc/bc250-smu-oc.conf") as f:
            m = _re.search(r"frequency\s*=\s*(\d+)", f.read())
    except OSError:
        # no CPU overclock config yet: report no frequency below
        m = None
    scx_on = subprocess.run(["systemctl", "is-enabled", "skillfish-scx.path"], capture_output=True, text=True).stdout.strip() == "enabled"
    stato = {"tetto": int(g.get("freq_max", 0) or 0), "cpu": int(m.group(1)) if m else 0,
             "ventola": fan.get("preset", ""), "scx": scx_on}
    attivo = None
    for k, p in profili.items():
        if (stato["tetto"] == p.get("tetto") and stato["ventola"] == p.get("ventola")
                and stato["scx"] == bool(p.get("scx")) and (not p.get("cpu") or stato["cpu"] == p.get("cpu"))):
            attivo = k
    return {"ok": True, "profili": profili, "stato": stato, "attivo": attivo}


def cc_profilo_applica(chiave):
    import re as _re
    p = cc_profili()["profili"].get(chiave)
    if not p:
        return {"ok": False, "err": "profilo sconosciuto"}
    errori = []
    g = _cc_conf_governor()
    pts = g.get("curva") or []
    if pts and p.get("tetto"):
        g["freq_max"] = max(pts[0][0], min(pts[-1][0], int(p["tetto"])))
        r = tuner_cmd([{"cmd": "gov-set", "conf": g}], timeout=90)
        if not r.get("ok"):
            errori.append("GPU: " + str(r.get("err", "?")))
    if p.get("cpu"):
        try:
            with open("/etc/bc250-smu-oc.conf") as f:
                t = f.read()
        except OSError:
            t = ""
        sc = _re.search(r"scale\s*=\s*(-?\d+)", t)
        tp = _re.search(r"max_temperature\s*=\s*(\d+)", t)
        r = tuner_cmd([{"cmd": "apply-cpu", "mhz": int(p["cpu"]), "scale": int(sc.group(1)) if sc else 0,
                        "temp": int(tp.group(1)) if tp else 85}], timeout=90)
        if not r.get("ok"):
            errori.append("CPU: " + str(r.get("err", "?")))
    if p.get("ventola") in CC_PRESET_VENTOLA:
        c = ventola_conf() or {}
        if c.get("pwm") and c.get("sorgente"):
            c["curva"] = CC_PRESET_VENTOLA[p["ventola"]]
            c["preset"] = p["ventola"]
            c["attivo"] = True
            r = ventola_scrivi(c)
            if not r.get("ok"):
                errori.append("ventola: " + str(r.get("error", r.get("err", "?"))))
        else:
            errori.append("ventola: configurala prima una volta dalla sua pagina")
    if "scx" in p:
        r = tuner_cmd([{"cmd": "scx-set", "on": bool(p["scx"])}], timeout=60)
        if not r.get("ok") and bool(p["scx"]):
            errori.append("scx: " + str(r.get("err", "?")))
    return {"ok": not errori, "err": "; ".join(errori), "profilo": chiave}


def launch_app(cmd):
    u = desktop_user()
    try:
        subprocess.Popen(["sudo", "-u", u, "env"] + ["%s=%s" % (k, v) for k, v in user_env().items()]
                         + ["setsid", "nohup"] + cmd,
                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": str(e)}


# ---------------- Hub (full app store: apt + flatpak + snap) ----------------
_HUBJOB = {"running": False, "title": "", "log": [], "done": True, "rc": 0}
_HUBLOCK = threading.Lock()
_PKG_RE = re.compile(r"^[a-z0-9][a-z0-9.+-]{0,80}$")          # apt
_FLAT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,120}$")  # flatpak app id
_SNAP_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,80}$")           # snap name
HUB_CATALOG = "/usr/local/bin/skillfish-hub-catalog"
_ICON_ROOTS = ["/usr/share/app-info/icons", "/var/lib/app-info/icons", "/usr/share/icons",
               "/usr/share/pixmaps", "/var/lib/flatpak"]
# In-memory app catalogue, built in the background from skillfish-hub-catalog.
HUBCAT = {"apps": [], "by_key": {}, "ready": False, "building": False, "ts": 0}
_CARD = ("key", "backend", "id", "pkgid", "name", "summary", "icon",
         "top", "rating", "rating_n", "installed", "ver", "developer", "ours")


def _apt_env():
    e = dict(os.environ); e["DEBIAN_FRONTEND"] = "noninteractive"; return e


def _card(a):
    c = {k: a.get(k) for k in _CARD}
    c["iloc"] = bool(a.get("icon_local"))
    return c


def _cat_build():
    if HUBCAT["building"]:
        return
    HUBCAT["building"] = True
    try:
        p = subprocess.run([HUB_CATALOG, "build"], capture_output=True, text=True, timeout=180)
        apps = json.loads(p.stdout).get("apps", [])
        HUBCAT["apps"] = apps
        HUBCAT["by_key"] = {a["key"]: a for a in apps}
        HUBCAT["ready"] = True
        HUBCAT["ts"] = int(time.time())
    except Exception as e:
        sys.stderr.write("hub catalog build failed: %s\n" % e)
    finally:
        HUBCAT["building"] = False


def cat_build_async():
    threading.Thread(target=_cat_build, daemon=True).start()


def hub_status():
    return {"ok": True, "ready": HUBCAT["ready"], "building": HUBCAT["building"],
            "count": len(HUBCAT["apps"]), "ts": HUBCAT["ts"]}


def hub_catalog(p):
    if not HUBCAT["ready"]:
        if not HUBCAT["building"]:
            cat_build_async()
        return {"ok": True, "ready": False, "apps": [], "total": 0}
    cat = p.get("cat"); sub = p.get("sub"); bk = p.get("backend")
    q = (p.get("q") or "").lower().strip(); sort = p.get("sort", "name")
    ours_only = p.get("ours") in ("1", "true", "yes")
    try:
        page = max(0, int(p.get("page", 0)))
    except Exception:
        page = 0
    per = 60
    res = []
    for a in HUBCAT["apps"]:
        if ours_only and not a.get("ours"):
            continue
        if cat and a.get("top") != cat:
            continue
        if sub and sub not in (a.get("cats") or []):
            continue
        if bk and a.get("backend") != bk:
            continue
        if q and q not in a.get("name", "").lower() and q not in (a.get("summary") or "").lower():
            continue
        res.append(a)
    if sort == "rating":
        res.sort(key=lambda a: (a.get("rating_n", 0), a.get("rating", 0)), reverse=True)
    elif sort == "installed":
        res.sort(key=lambda a: (not a.get("installed"), a.get("name", "").lower()))
    else:
        res.sort(key=lambda a: a.get("name", "").lower())
    total = len(res)
    cards = [_card(a) for a in res[page * per:(page + 1) * per]]
    return {"ok": True, "ready": True, "total": total, "page": page, "per": per, "apps": cards}


def hub_categories():
    counts = {}
    for a in HUBCAT["apps"]:
        counts[a.get("top")] = counts.get(a.get("top"), 0) + 1
    return {"ok": True, "ready": HUBCAT["ready"], "counts": counts, "total": len(HUBCAT["apps"])}


def hub_app(key):
    a = HUBCAT["by_key"].get(key)
    if not a:
        return {"ok": False, "error": "app non trovata"}
    a = dict(a)
    a["iloc"] = bool(a.get("icon_local"))
    try:
        p = subprocess.run([HUB_CATALOG, "reviews", a.get("id", ""), a.get("ver", "")],
                           capture_output=True, text=True, timeout=20)
        a["reviews"] = json.loads(p.stdout)
    except Exception:
        a["reviews"] = []
    return {"ok": True, "app": a}


def hub_icon_path(key):
    a = HUBCAT["by_key"].get(key)
    if not a:
        return None
    p = a.get("icon_local")
    if not p:
        return None
    rp = os.path.realpath(p)
    if not any(rp == os.path.realpath(r) or rp.startswith(os.path.realpath(r) + os.sep) for r in _ICON_ROOTS):
        return None
    return rp if os.path.isfile(rp) else None


def hub_search(q):
    q = (q or "").strip()
    if len(q) < 2:
        return {"ok": True, "results": []}
    ql = q.lower()
    res = [a for a in HUBCAT["apps"] if a.get("backend") in ("apt", "flatpak")
           and (ql in a.get("name", "").lower() or ql in (a.get("summary") or "").lower()
                or ql in a.get("pkgid", "").lower())][:80]
    snaps = []
    try:
        p = subprocess.run([HUB_CATALOG, "snap-find", q], capture_output=True, text=True, timeout=25)
        snaps = json.loads(p.stdout)
    except Exception:
        snaps = []
    return {"ok": True, "results": [_card(a) for a in res] + [_card(s) for s in snaps]}


def hub_installed():
    res = [_card(a) for a in HUBCAT["apps"] if a.get("installed")]
    res.sort(key=lambda a: a.get("name", "").lower())
    return {"ok": True, "installed": res, "count": len(res)}


def hub_updates():
    ups = []
    try:
        out = subprocess.run(["apt-get", "-s", "full-upgrade"], capture_output=True,
                             text=True, timeout=60, env=_apt_env()).stdout
        for ln in out.splitlines():
            m = re.match(r"Inst (\S+) \[([^\]]*)\] \(([^ ]+)", ln)
            if m:
                ups.append({"backend": "apt", "pkg": m.group(1), "old": m.group(2), "new": m.group(3)})
    except Exception:
        pass
    try:
        out = subprocess.run(["flatpak", "remote-ls", "--updates", "--columns=application,version"],
                             capture_output=True, text=True, timeout=40).stdout
        for ln in out.splitlines():
            f = ln.split("\t")
            if f and f[0] and f[0] != "Application ID":
                ups.append({"backend": "flatpak", "pkg": f[0], "old": "", "new": f[1] if len(f) > 1 else ""})
    except Exception:
        pass
    try:
        out = subprocess.run(["snap", "refresh", "--list"], capture_output=True, text=True, timeout=40).stdout
        for ln in out.splitlines()[1:]:
            f = ln.split()
            if f:
                ups.append({"backend": "snap", "pkg": f[0], "old": "", "new": f[1] if len(f) > 1 else ""})
    except Exception:
        pass
    return {"ok": True, "updates": ups, "count": len(ups)}


def hub_sources():
    d = "/etc/apt/sources.list.d"
    out = []
    try:
        for fn in sorted(os.listdir(d)):
            if not fn.endswith(".sources"):
                continue
            with open(os.path.join(d, fn)) as f:
                txt = f.read()
            en = not re.search(r"(?im)^Enabled:\s*no", txt)
            mu = re.search(r"(?im)^URIs:\s*(\S+)", txt)
            out.append({"name": fn[:-8], "enabled": en, "uri": mu.group(1) if mu else ""})
    except Exception:
        pass
    remotes = []
    try:
        o = subprocess.run(["flatpak", "remotes", "--columns=name,url"],
                           capture_output=True, text=True, timeout=15).stdout
        for ln in o.splitlines():
            f = ln.split("\t")
            if f and f[0]:
                remotes.append({"name": f[0].strip(), "uri": f[1].strip() if len(f) > 1 else ""})
    except Exception:
        pass
    return {"ok": True, "sources": out, "flatpak_remotes": remotes}


def _hub_run(title, cmds, stop_on_error=True):
    """Run a list of argv commands sequentially in a background thread; refresh catalogue after."""
    with _HUBLOCK:
        if _HUBJOB["running"]:
            return False
        _HUBJOB.update(running=True, title=title, log=[], done=False, rc=None)

    def worker():
        rc = 0
        try:
            for argv in cmds:
                p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                     text=True, env=_apt_env())
                for line in p.stdout:
                    _HUBJOB["log"].append(line.rstrip())
                    if len(_HUBJOB["log"]) > 600:
                        del _HUBJOB["log"][:150]
                p.wait()
                if p.returncode != 0:
                    rc = p.returncode
                    if stop_on_error:
                        break
        except Exception as e:
            _HUBJOB["log"].append("errore: %s" % e); rc = 1
        finally:
            _HUBJOB["rc"] = rc; _HUBJOB["running"] = False; _HUBJOB["done"] = True
            cat_build_async()  # refresh installed state

    threading.Thread(target=worker, daemon=True).start()
    return True


def hub_op(op, backend, pkg):
    pkg = (pkg or "").strip()
    if op == "update":
        cmds = [["apt-get", "update"]]
    elif op == "upgrade":
        cmds = [["apt-get", "update"], ["apt-get", "-y", "full-upgrade"],
                ["flatpak", "update", "--system", "-y", "--noninteractive"], ["snap", "refresh"]]
        return _hub_start("aggiorno tutto", cmds, stop_on_error=False)
    elif op in ("install", "remove"):
        if backend == "apt":
            if not _PKG_RE.match(pkg):
                return {"ok": False, "error": "nome pacchetto non valido"}
            cmds = ([["apt-get", "update"], ["apt-get", "install", "-y", pkg]] if op == "install"
                    else [["apt-get", "purge", "-y", pkg], ["apt-get", "autoremove", "-y"]])
        elif backend == "flatpak":
            if not _FLAT_RE.match(pkg):
                return {"ok": False, "error": "id flatpak non valido"}
            cmds = ([["flatpak", "install", "--system", "-y", "--noninteractive", "flathub", pkg]] if op == "install"
                    else [["flatpak", "uninstall", "--system", "-y", "--noninteractive", pkg]])
        elif backend == "snap":
            if not _SNAP_RE.match(pkg):
                return {"ok": False, "error": "nome snap non valido"}
            cmds = [["snap", "install", pkg]] if op == "install" else [["snap", "remove", pkg]]
        else:
            return {"ok": False, "error": "backend sconosciuto"}
    else:
        return {"ok": False, "error": "operazione sconosciuta"}
    return _hub_start(op + ((" " + pkg) if pkg else ""), cmds)


def _hub_start(title, cmds, stop_on_error=True):
    if not _hub_run(title, cmds, stop_on_error):
        return {"ok": False, "error": "operazione già in corso"}
    return {"ok": True, "started": True}


def hub_source_toggle(name, enable):
    if not re.match(r"^[A-Za-z0-9_-]+$", name or ""):  # no dots/slashes -> no traversal
        return {"ok": False, "error": "nome non valido"}
    root = "/etc/apt/sources.list.d"
    f = os.path.realpath(os.path.join(root, name + ".sources"))
    if os.path.dirname(f) != os.path.realpath(root):
        return {"ok": False, "error": "percorso non valido"}
    val = "yes" if enable else "no"
    try:
        with open(f) as fh:
            txt = fh.read()
        if re.search(r"(?im)^Enabled:", txt):
            txt = re.sub(r"(?im)^Enabled:.*$", "Enabled: " + val, txt)
        else:
            txt = txt.rstrip() + "\nEnabled: " + val + "\n"
        with open(f, "w") as fh:
            fh.write(txt)
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": str(e)}


def read_log(which, n=200):
    n = max(1, min(2000, int(n)))
    if which == "freeze":
        try:
            with open("/var/log/skillfish-freeze.log") as f:
                return f.read().splitlines()[-n:]
        except Exception:
            return []
    cmd = ["journalctl", "-n", str(n), "--no-pager", "-o", "short-iso"]
    if which == "kernel":
        cmd.append("-k")
    try:
        return subprocess.run(cmd, capture_output=True, text=True, timeout=5).stdout.splitlines()
    except Exception:
        return []


# (server-side benchmark recorder removed — recording lives in the native Telemetry app)


# ---------------- interactive services (KVM / Terminal) ----------------
import secrets
SVC_PROC = {}                       # name -> Popen
VNC_PASS = secrets.token_hex(4)     # 8 hex chars, regenerated each daemon start
TTYD_TOKEN = secrets.token_urlsafe(12)
KVM_PORT, TTYD_PORT = 6080, 7681


def _alive(name):
    p = SVC_PROC.get(name)
    return p is not None and p.poll() is None


def _spawn(name, argv, as_user=False, kill_pat=None):
    if _alive(name):
        return
    # a previous run may have left an orphan squatting the port → clear it first
    if kill_pat:
        subprocess.run(["pkill", "-f", kill_pat], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        time.sleep(0.3)
    if as_user:
        u = desktop_user()
        argv = ["sudo", "-u", u, "env"] + ["%s=%s" % (k, v) for k, v in user_env().items()] + argv
    SVC_PROC[name] = subprocess.Popen(argv, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def kvm_start():
    # x11vnc on a dedicated port (5901), localhost + password; websockify bridges it to
    # noVNC on localhost:KVM_PORT (plain) — the dashboard reverse-proxies it under /kvm
    # over its own TLS + session, so there's no second cert/login and no LAN exposure.
    _spawn("x11vnc", ["x11vnc", "-display", ":0", "-rfbport", "5901", "-localhost",
                      "-passwd", VNC_PASS, "-forever", "-shared", "-noxdamage", "-repeat", "-quiet"],
           as_user=True, kill_pat="x11vnc -display :0 -rfbport 5901")
    time.sleep(1.0)
    _spawn("websockify", ["websockify", "--web", "/usr/share/novnc",
                          "127.0.0.1:%d" % KVM_PORT, "localhost:5901"],
           kill_pat="websockify --web")
    time.sleep(0.8)
    return {"ok": True, "password": VNC_PASS}


def kvm_stop():
    for n in ("websockify", "x11vnc"):
        p = SVC_PROC.pop(n, None)
        if p and p.poll() is None:
            p.terminate()
    return {"ok": True}


def terminal_start():
    u = desktop_user()
    # ttyd on loopback only, base-path /terminal, NO auth/TLS of its own — the dashboard
    # reverse-proxies it under /terminal (its TLS + login session = single sign-on).
    # Command after `--` so ttyd's getopt can't eat the shell's flags.
    _spawn("ttyd", ["ttyd", "-i", "lo", "-p", str(TTYD_PORT), "-b", "/terminal",
                    "-W", "-t", "fontSize=14", "--", "su", "-", u],
           kill_pat="ttyd -i lo")
    time.sleep(0.8)
    return {"ok": True}


def terminal_stop():
    p = SVC_PROC.pop("ttyd", None)
    if p and p.poll() is None:
        p.terminate()
    return {"ok": True}


# ---------------- AI ----------------
# Two engines are supported. Unsloth Studio is the current one: a native service
# running GGUF models through llama.cpp's Vulkan backend, which is the only
# GPU-accelerated path on the BC-250 (gfx1013 has no ROCm). It serves both its own
# chat UI and an OpenAI-compatible API on one port.
#
# Qui c'era anche il vecchio stack Ollama + Open WebUI su Docker, pilotato con
# `docker compose`. Docker e' stato rimosso dal sistema (non serviva piu' a
# niente: girava un solo container, dockge, che gestiva zero stack), quindi
# quelle strade erano diventate impercorribili.
UNSLOTH_PORT = 8888
UNSLOTH_SVC = "skillfish-unsloth.service"


def _port_open(port):
    try:
        s = socket.create_connection(("127.0.0.1", port), timeout=1); s.close(); return True
    except Exception:
        return False


def _unsloth_installed():
    return os.path.exists("/usr/local/bin/skillfish-unsloth")


def ai_engine():
    """Il motore AI di SkillFishOS: uno solo, Unsloth Studio.

    Resta una funzione invece di una costante perche' il frontend legge il campo
    "engine" e domani il motore potrebbe cambiare ancora.
    """
    return "unsloth"


UNSLOTH_BOOTSTRAP = "/root/.unsloth/studio/auth/.bootstrap_password"


def _unsloth_first_login():
    """Utente e password iniziali di Unsloth, finche' non li ha cambiati.

    Unsloth NON ha una password predefinita da pubblicare: ne genera una a caso
    a ogni installazione e la scrive in un file leggibile dal solo root,
    annunciandola una volta sola in un log che nessuno legge. Chi apre Studio si
    trova davanti a una richiesta di accesso e finisce a cercare su internet una
    risposta che non esiste.

    E c'e' una scadenza: se la password non viene cambiata, Unsloth si spegne da
    solo dopo un'ora. Chi non sa nemmeno di doverla cambiare vede il motore AI
    morire senza motivo apparente.

    Quindi la mostriamo qui, dietro il login PAM della dashboard, e sparisce da
    sola nel momento in cui l'utente la cambia: `requires_password_change`
    diventa falso e non abbiamo piu' niente da dire.
    """
    try:
        req = urllib.request.Request("http://127.0.0.1:%d/api/auth/status" % UNSLOTH_PORT)
        st = json.loads(urllib.request.urlopen(req, timeout=4).read().decode())
    except Exception:
        return {}
    if not st.get("requires_password_change"):
        return {"user": st.get("default_username", "")}
    try:
        with open(_unsloth_bootstrap()) as f:
            pw = f.read().strip()
    except Exception:
        pw = ""
    return {"user": st.get("default_username", "unsloth"), "password": pw,
            "must_change": True}


def _unsloth_models():
    """I modelli che Unsloth ha caricato, chiesti alla sua API.

    Prima restituivamo sempre una lista vuota, con la motivazione che i modelli
    stanno dietro l'autenticazione di Unsloth. Vero, ma la conseguenza era un
    menu a tendina vuoto nella chat: l'utente non poteva scegliere niente.
    Ora che la chiave API si imposta dalla dashboard, quell'autenticazione ce
    l'abbiamo e possiamo semplicemente chiedere.

    Senza chiave restituisce comunque una lista vuota, che e' corretto: non
    sapremmo nemmeno se il motore ha dei modelli.
    """
    key = _unsloth_key()
    if not key or not _port_open(UNSLOTH_PORT):
        return []
    try:
        req = urllib.request.Request("http://127.0.0.1:%d/v1/models" % UNSLOTH_PORT,
                                     headers={"Authorization": "Bearer " + key})
        out = json.loads(urllib.request.urlopen(req, timeout=8).read().decode())
        return sorted(m.get("id", "") for m in out.get("data", []) if m.get("id"))
    except Exception:
        return []


# ---- Unsloth Studio through its own API -----------------------------------------
UNSLOTH_DEFAULT_F = "/etc/default/skillfish-unsloth"
UNSLOTH_UPDATE_BIN = "/usr/local/bin/skillfish-unsloth-update"


def _unsloth_home():
    """The home of the user Unsloth is installed for: the dashboard user, else
    the owner of the graphical session, else the first real user with ~/.unsloth."""
    import pwd
    candidati = [desktop_user()]
    try:
        out = subprocess.run(["loginctl", "list-sessions", "--no-legend"], capture_output=True,
                             text=True, timeout=5).stdout
        candidati += [ln.split()[2] for ln in out.splitlines() if len(ln.split()) > 2]
    except (OSError, subprocess.SubprocessError):
        # no logind session to ask: fall back to the other candidates below
        pass
    candidati += [p.pw_name for p in pwd.getpwall() if 1000 <= p.pw_uid < 65534]
    for u in candidati:
        if not u or u == "root":
            continue
        try:
            h = pwd.getpwnam(u).pw_dir
        except KeyError:
            continue
        if os.path.isdir(os.path.join(h, ".unsloth")):
            return h
    return ""


def _unsloth_bootstrap():
    h = _unsloth_home()
    return os.path.join(h, ".unsloth", "studio", "auth", ".bootstrap_password") if h else UNSLOTH_BOOTSTRAP


def _studio(path, data=None, key=None, t=8, method=None):
    """One call to Studio's API; JSON in, JSON out, None when unreachable."""
    hdr = {"Accept": "application/json"}
    if key:
        hdr["Authorization"] = "Bearer " + key
    body = None
    if data is not None:
        body = json.dumps(data).encode()
        hdr["Content-Type"] = "application/json"
    req = urllib.request.Request("http://127.0.0.1:%d%s" % (UNSLOTH_PORT, path), data=body, headers=hdr,
                                 method=method or ("POST" if data is not None else "GET"))
    try:
        with urllib.request.urlopen(req, timeout=t) as r:
            return json.loads(r.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        try:
            return {"_error": e.code, **json.loads(e.read().decode())}
        except Exception:
            return {"_error": e.code}
    except Exception:
        return None


_AI_VER = {"t": 0.0, "v": {}}


def _unsloth_version():
    """Version, latest and backend, from Studio when the key allows it; cached
    ten minutes because the update check goes to the network."""
    key = _unsloth_key()
    if not key or not _port_open(UNSLOTH_PORT):
        return {}
    if time.time() - _AI_VER["t"] < 600 and _AI_VER["v"]:
        return _AI_VER["v"]
    agg = _studio("/api/studio/update-status", key=key, t=10) or {}
    be = _studio("/api/llama/backend", key=key, t=6) or {}
    v = {}
    if agg.get("current_version"):
        v = {"version": agg.get("current_version"), "latest": agg.get("latest_version"),
             "update_available": bool(agg.get("update_available")),
             "backend": (be.get("backend") or "") + ((" " + be["installed_tag"]) if be.get("installed_tag") else "")}
        _AI_VER.update(t=time.time(), v=v)
    return v


def _unsloth_conf_read():
    out = {"bind": "127.0.0.1", "parallel": 4}
    try:
        with open(UNSLOTH_DEFAULT_F) as f:
            for line in f:
                line = line.strip()
                if line.startswith("UNSLOTH_BIND="):
                    out["bind"] = line.split("=", 1)[1].strip().strip('"')
                elif line.startswith("UNSLOTH_PARALLEL="):
                    try:
                        out["parallel"] = int(line.split("=", 1)[1].strip().strip('"'))
                    except ValueError:
                        # malformed value: fall back to the default set above
                        out["parallel"] = 4
    except OSError:
        # no environment file yet: report the defaults set above
        pass
    out["lan"] = out["bind"] not in ("127.0.0.1", "localhost", "::1")
    r = subprocess.run(["systemctl", "is-enabled", UNSLOTH_SVC], capture_output=True, text=True)
    out["autostart"] = r.stdout.strip() == "enabled"
    return out


def _ip_locale():
    """L'indirizzo della scheda sulla rete di casa.

    ⚠️ Si apre un socket UDP verso un indirizzo qualsiasi e si guarda da quale
    interfaccia sarebbe uscito: non si manda niente e non serve che quel
    destinatario esista. Leggere `hostname -I` darebbe anche gli indirizzi di
    docker e di ZeroTier, e non si saprebbe quale scegliere.
    """
    import socket
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("192.0.2.1", 9))          # TEST-NET-1: non esiste apposta
        return s.getsockname()[0]
    except OSError:
        return "127.0.0.1"
    finally:
        s.close()


CLUSTER_STATO = "/run/skillfish/cluster.json"


def cluster_stato():
    """Lo stato del cluster, letto dal file che scrive il servizio.

    ⚠️ Si legge e basta: niente ssh, niente root. La raccolta la fa
    skillfish-cluster.service, che e' l'unico che ha la chiave delle altre
    schede.
    """
    try:
        with open(CLUSTER_STATO, encoding="utf-8") as f:
            d = json.load(f)
    except (OSError, ValueError):
        # Il servizio puo' essere spento: si risponde "nessuna scheda" invece
        # di un errore, perche' per chi guarda e' la stessa cosa.
        return {"ok": True, "schede": [], "memoria_totale": 0,
                "memoria_usata": 0, "watt_totali": 0, "servizio": False}
    d["ok"] = True
    d["servizio"] = True
    # ⚠️ Il file sopravvive al servizio che lo scrive: se e' vecchio si dice,
    # invece di servire l'ultima fotografia come se fosse adesso.
    ag = d.get("aggiornato") or 0
    d["eta"] = max(0, int(time.time() - ag)) if ag else 0
    d["fermo"] = d["eta"] > 30
    return d


LLAMA = "/usr/local/bin/skillfish-llama"


def ai_llama():
    """Il livello scelto e i modelli sul disco, per la tendina della pagina."""
    if not os.path.exists(LLAMA):
        return {}
    try:
        c = json.loads(subprocess.run([LLAMA, "conf"], capture_output=True,
                                      text=True, timeout=15).stdout or "{}")
        m = json.loads(subprocess.run([LLAMA, "modelli"], capture_output=True,
                                      text=True, timeout=30).stdout or "{}")
    except (ValueError, OSError, subprocess.SubprocessError):
        return {}
    return {"livello": c.get("livello") or "studio",
            "modello": c.get("modello") or "",
            "modelli": m.get("modelli") or []}


# I tre livelli della modalita' AI, scritti una volta sola: chi arriva da fuori
# non aggiunge niente a questo elenco, al massimo ne indica uno.
LIVELLI_AI = ("studio", "motore", "motore-api")


def ai_livello(livello="", modello=""):
    """Cambia la scelta. ⚠️ Non accende e non spegne niente: accendere e'
    un'altra rotta, con la sua domanda."""
    # ⚠️ I DUE VALORI ARRIVANO DAL BROWSER E NON ENTRANO NELLA RIGA DI COMANDO.
    # Servono soltanto a scegliere: il livello fra i tre che conosciamo, il
    # modello fra i file che skillfish-llama ha trovato sul disco. Quello che
    # passiamo e' sempre l'elemento preso da quegli elenchi, cioe' roba nostra.
    # Confrontare e poi passare comunque la stringa di chi chiama sembrava
    # equivalente: non lo e', perche' basta allentare il confronto un giorno e
    # il percorso torna ad arrivare da fuori.
    cmd = [LLAMA, "imposta"]
    if livello:
        scelto = ""
        for noto in LIVELLI_AI:
            if noto == livello:
                scelto = noto
                break
        if not scelto:
            return {"ok": False, "errore": "livello sconosciuto"}
        cmd.append("livello=" + scelto)
    if modello:
        scelto = ""
        for m in (ai_llama().get("modelli") or []):
            if m.get("file") == modello:
                scelto = m.get("file") or ""
                break
        if not scelto:
            return {"ok": False, "errore": "modello sconosciuto"}
        cmd.append("modello=" + scelto)
    if len(cmd) == 2:
        return {"ok": False, "errore": "niente da cambiare"}
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    try:
        return json.loads(r.stdout or "{}")
    except ValueError:
        return {"ok": False, "errore": (r.stderr or r.stdout or "").strip()[-200:]}


def ai_mode(azione):
    """La modalita' AI, vista da remoto.

    ⚠️ E' QUI CHE SI TORNA INDIETRO. Con il desktop spento il Control Center
    non esiste, quindi questa rotta e' l'unica strada che non passa da ssh: il
    demone della dashboard non si ferma mai, proprio per questo.
    """
    if azione not in ("on", "off", "stato"):
        return {"ok": False, "errore": "azione sconosciuta"}
    r = subprocess.run(["/usr/local/bin/skillfish-ai-mode", azione],
                       capture_output=True, text=True, timeout=120)
    try:
        return {"ok": r.returncode == 0, **json.loads(r.stdout or "{}")}
    except ValueError:
        return {"ok": False, "errore": (r.stderr or r.stdout or "").strip()[-300:]}


def ai_conf():
    conf = _unsloth_conf_read()
    # L'indirizzo da incollare in un editor o in opencode. Lo diciamo noi
    # perche' il browser vede l'indirizzo da cui ha aperto la pagina, che puo'
    # essere un nome, ZeroTier o un inoltro di porta: quello che serve e'
    # l'indirizzo della SCHEDA sulla rete di casa.
    conf["indirizzo"] = "http://%s:%d" % (
        _ip_locale() if conf.get("lan") else "127.0.0.1", conf.get("porta", 8888))
    return {"ok": True, **conf}


def ai_conf_set(d):
    cur = _unsloth_conf_read()
    lan = bool(d.get("lan", cur["lan"]))
    try:
        parallel = max(1, min(64, int(d.get("parallel", cur["parallel"]))))
    except (TypeError, ValueError):
        parallel = cur["parallel"]
    try:
        with open(UNSLOTH_DEFAULT_F, "w") as f:
            f.write("# SkillFishOS: read by skillfish-unsloth.service. Written by the Control Center\n"
                    "# and the Remote Manager; edit by hand if you like, then restart the unit.\n"
                    "UNSLOTH_BIND=%s\nUNSLOTH_PARALLEL=%d\n" % ("0.0.0.0" if lan else "127.0.0.1", parallel))
    except OSError as e:
        return {"ok": False, "error": str(e)}
    if "autostart" in d:
        subprocess.run(["systemctl", "enable" if d["autostart"] else "disable", UNSLOTH_SVC],
                       capture_output=True, text=True, timeout=60)
    if (lan, parallel) != (cur["lan"], cur["parallel"]) and _port_open(UNSLOTH_PORT):
        subprocess.Popen(["systemctl", "restart", UNSLOTH_SVC], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return ai_conf()


_AI_UPD = {"running": False, "log": [], "rc": None, "t": 0.0}


def ai_update_start():
    """skillfish-unsloth-update in a thread; the log is read back by /api/ai/update."""
    if _AI_UPD["running"]:
        return {"ok": True, "running": True}
    if not os.path.exists(UNSLOTH_UPDATE_BIN):
        return {"ok": False, "error": "skillfish-unsloth-update missing"}

    def worker():
        _AI_UPD.update(running=True, log=[], rc=None, t=time.time())
        try:
            p = subprocess.Popen([UNSLOTH_UPDATE_BIN], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                 text=True, bufsize=1)
            for ln in iter(p.stdout.readline, ""):
                _AI_UPD["log"].append(ln.rstrip()[:300])
                del _AI_UPD["log"][:-400]
            p.wait()
            _AI_UPD["rc"] = p.returncode
        except Exception as e:
            _AI_UPD["log"].append(str(e))
            _AI_UPD["rc"] = 1
        _AI_VER["t"] = 0.0
        _AI_UPD["running"] = False
    threading.Thread(target=worker, daemon=True).start()
    return {"ok": True, "running": True}


def ai_update_state():
    return {"ok": True, "running": _AI_UPD["running"], "rc": _AI_UPD["rc"],
            "log": "\n".join(_AI_UPD["log"][-80:])}


def _unsloth_mint_key(password):
    r = _studio("/api/auth/login", {"username": "unsloth", "password": password})
    if not r or not r.get("access_token"):
        return None, "Studio password not accepted"
    tok = r["access_token"]
    r = _studio("/api/auth/api-keys", {"name": "SkillFishOS Remote Manager"}, key=tok)
    if not r or not r.get("key"):
        return None, "Studio did not issue the key"
    # a request for a model that is not loaded must load it, not fail
    _studio("/api/settings/openai-auto-switch", {"enabled": True}, key=tok, method="PUT")
    return r["key"], ""


def ai_key_create(password):
    if not password:
        return {"ok": False, "error": "password missing"}
    k, err = _unsloth_mint_key(password)
    if not k:
        return {"ok": False, "error": err}
    salva_chiave_unsloth(k)
    _AI_VER["t"] = 0.0
    return {"ok": True, "has_key": True}


def ai_setup(new_password):
    """The first sign-in done for the user: bootstrap password in, their
    password set, a key minted and saved."""
    if not new_password or len(new_password) < 8:
        return {"ok": False, "error": "password too short (8+)"}
    try:
        with open(_unsloth_bootstrap()) as f:
            old = f.read().strip()
    except OSError:
        return {"ok": False, "error": "initial password not found"}
    r = _studio("/api/auth/login", {"username": "unsloth", "password": old})
    if not r or not r.get("access_token"):
        return {"ok": False, "error": "the initial password is no longer valid: sign in to Studio"}
    r2 = _studio("/api/auth/change-password", {"current_password": old, "new_password": new_password},
                 key=r["access_token"])
    if not r2 or r2.get("_error"):
        return {"ok": False, "error": (r2 or {}).get("detail") or "password change refused"}
    return ai_key_create(new_password)


def ai_variants(repo):
    key = _unsloth_key()
    if not key or not repo or "/" not in repo:
        return {"ok": False, "error": "key or repository missing"}
    # un repository di Hugging Face e' utente/nome: due pezzi, niente altro.
    # Il valore arriva dal browser e finisce dentro un indirizzo: si controlla
    # la forma prima, non ci si affida a quote()
    if not re.match(r"^[A-Za-z0-9._-]{1,80}/[A-Za-z0-9._-]{1,120}$", repo):
        return {"ok": False, "error": "nome del repository non valido"}
    r = _studio("/api/hub/gguf-variants?repo_id=" + urllib.parse.quote(repo, safe=""), key=key, t=40)
    if not r or r.get("_error"):
        return {"ok": False, "error": (r or {}).get("detail") or "no answer"}
    return {"ok": True, "variants": [{"quant": v.get("quant"), "filename": v.get("filename"),
                                      "size_bytes": v.get("size_bytes"), "downloaded": bool(v.get("downloaded"))}
                                     for v in r.get("variants") or []]}


def ai_download(repo, variant):
    key = _unsloth_key()
    if not key or not repo:
        return {"ok": False, "error": "key or repository missing"}
    if not re.match(r"^[A-Za-z0-9._-]{1,80}/[A-Za-z0-9._-]{1,120}$", repo):
        return {"ok": False, "error": "nome del repository non valido"}
    if variant is not None and not re.match(r"^[A-Za-z0-9._-]{1,60}$", str(variant)):
        return {"ok": False, "error": "variante non valida"}
    r = _studio("/api/hub/download", {"repo_id": repo, "gguf_variant": variant}, key=key, t=40)
    if not r or r.get("_error") or not (r.get("accepted") or r.get("state") == "running"):
        return {"ok": False, "error": (r or {}).get("detail") or "Studio did not accept the download"}
    return {"ok": True, "job": r.get("job_key")}


def ai_downloads():
    key = _unsloth_key()
    if not key or not _port_open(UNSLOTH_PORT):
        return {"ok": True, "downloads": []}
    att = (_studio("/api/hub/active-downloads", key=key, t=6) or {}).get("downloads") or []
    out = []
    for d in att:
        repo = d.get("repo_id") or ""
        if not re.match(r"^[A-Za-z0-9._-]{1,80}/[A-Za-z0-9._-]{1,120}$", repo or ""):
            continue
        prog = _studio("/api/hub/download-progress?repo_id=" + urllib.parse.quote(repo, safe=""), key=key, t=6) or {}
        tot = prog.get("expected_bytes") or prog.get("total_bytes") or 0
        done = prog.get("downloaded_bytes") or 0
        pct = int(100.0 * done / tot) if tot else None
        out.append({"repo_id": repo, "variant": d.get("variant"), "state": d.get("state"),
                    "percent": pct, "downloaded_bytes": done, "total_bytes": tot})
    return {"ok": True, "downloads": out}


def _unsloth_cached():
    key = _unsloth_key()
    if not key or not _port_open(UNSLOTH_PORT):
        return []
    r = _studio("/v1/models", key=key, t=6) or {}
    return [{"id": m.get("id"), "quant": m.get("quant"), "loaded": bool(m.get("loaded"))}
            for m in r.get("data") or [] if m.get("id")]


def ai_status():
    running = _port_open(UNSLOTH_PORT)
    return {"ok": True, "engine": "unsloth", "running": running,
            "webui": running, "webui_path": "/unsloth", "port": UNSLOTH_PORT,
            "first_login": _unsloth_first_login() if running else {},
            # solo se c'e' o non c'e': la chiave non esce MAI da qui
            "has_key": bool(_unsloth_key()),
            "models": _unsloth_models(),
            "catalog": _unsloth_cached() if running else [],
            "installed": _unsloth_installed() and bool(_unsloth_home()),
            "conf": _unsloth_conf_read(),
            "update_running": _AI_UPD["running"],
            # ⚠️ Il segno lo scrive skillfish-ai-mode: si guarda quello e non
            # si lancia un comando a ogni giro del riquadro.
            "ai_mode": os.path.exists("/var/lib/skillfish/ai-mode"),
            **ai_llama(),
            **(_unsloth_version() if running else {})}


def ai_start():
    subprocess.Popen(["systemctl", "start", UNSLOTH_SVC],
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return {"ok": True}


def ai_stop():
    subprocess.Popen(["systemctl", "stop", UNSLOTH_SVC],
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return {"ok": True}


def ai_pull(model):
    # Con Ollama il modello lo scaricava la dashboard. Unsloth Studio ha il
    # proprio catalogo dietro la propria autenticazione: da qui non possiamo
    # scaricarlo per conto dell'utente, e fingere di poterlo fare sarebbe peggio
    # che dirglielo.
    return {"ok": False,
            "error": "I modelli si scaricano da Unsloth Studio: apri «Apri Unsloth Studio» "
                     "e scegli il modello da li'."}


# ---- "Optimize for AI" (#5): tune GTT cap, KV cache and context for bigger models ----
GRUB_F = "/etc/default/grub"


def _cmdline_param(name):
    try:
        with open("/proc/cmdline") as f:
            for tok in f.read().split():
                if tok.startswith(name + "="):
                    return tok.split("=", 1)[1]
    except Exception:
        pass
    return None


GTT_PREDEFINITO_MB = 6144


def _ram_mb():
    try:
        with open("/proc/meminfo") as f:
            m = re.search(r"MemTotal:\s+(\d+)", f.read())
            return int(m.group(1)) // 1024 if m else 0
    except Exception:
        return 0


def _gtt_cap_mb():
    """Il tetto del GTT che comanda davvero, in MB.

    Stessa regola del driver: amdgpu.gttsize se c'e' e non e' -1 (deprecato dal
    kernel 7.x), altrimenti ttm.pages_limit, che sono pagine da 4 KiB. Prima si
    leggeva solo il primo, quindi su un sistema aggiornato la dashboard diceva
    che il GTT non era configurato mentre invece lo era.
    """
    v = _cmdline_param("amdgpu.gttsize")
    if (v or "").isdigit():
        return int(v)
    v = _cmdline_param("ttm.pages_limit")
    if (v or "").isdigit():
        return int(v) // 256
    return None


def ai_tune_status():
    gtt_cap = _gtt_cap_mb()
    swap_mb = 0
    try:
        with open("/proc/meminfo") as f:
            m = re.search(r"SwapTotal:\s+(\d+)", f.read())
            swap_mb = int(m.group(1)) // 1024 if m else 0
    except Exception:
        pass
    # KV-cache e lunghezza del contesto erano variabili del compose di Ollama.
    # In Unsloth quei parametri stanno dentro Studio e non li tocchiamo da qui:
    # restano le leve di sistema, il cap della GTT e lo swap. I campi vuoti li
    # teniamo perche' il frontend li legge.
    # Si consiglia di alzare il GTT solo se c'e' davvero margine da guadagnare:
    # sotto il mezzo giga non vale un riavvio. Prima lo si consigliava sempre,
    # anche quando il tetto era gia' al massimo che la RAM consente.
    ram = _ram_mb()
    margine = (ram - 1024 - gtt_cap) if (ram and gtt_cap) else 0
    recs = ["gtt_unlock"] if margine >= 512 else []
    return {"ok": True, "engine": "unsloth",
            "gtt_cap_mb": gtt_cap,
            "swap_mb": swap_mb, "kv_cache": "", "context": "",
            # Con Unsloth la GPU la usa llama.cpp col backend Vulkan.
            "vulkan": True, "flash_attention": None,
            "recommend": recs}


def _grub_update():
    subprocess.run(["update-grub"], capture_output=True, text=True, timeout=60)


def ai_tune_apply(action):
    try:
        if action in ("kv_q8", "kv_f16"):
            return {"ok": False, "error": "KV cache: si imposta dentro Unsloth Studio"}
        if action == "context":
            return {"ok": False, "error": "non implementato qui"}
        if action in ("gtt_unlock", "gtt_restore"):
            # ⚠️ Prima "unlock" toglieva amdgpu.gttsize e basta. Con
            # ttm.pages_limit=4194304 in riga di avvio quel gesto non liberava
            # niente: portava il tetto del GTT a 16 GB su una scheda che di RAM
            # ne ha 7,5, cioe' toglieva la rete invece di dare memoria. Adesso
            # si passa da skillfish-gtt, che scrive una leva sola, si rifiuta di
            # superare la RAM e rimette la riga di prima se non torna.
            ram = _ram_mb()
            mb = (ram - 1024) if action == "gtt_unlock" else GTT_PREDEFINITO_MB
            if mb < 512:
                return {"ok": False, "error": "RAM insufficiente per il GTT"}
            r = subprocess.run(["skillfish-gtt", str(mb)],
                               capture_output=True, text=True, timeout=180)
            if r.returncode != 0:
                return {"ok": False,
                        "error": (r.stderr or r.stdout or "skillfish-gtt").strip()[:300]}
            return {"ok": True, "reboot_needed": True,
                    "note": "Modifica al boot applicata: riavvia per attivarla."}
        return {"ok": False, "error": "azione sconosciuta"}
    except Exception as e:
        return {"ok": False, "error": str(e)}


def _unsloth_key():
    """API key for Unsloth's OpenAI-compatible endpoint, if the user configured one.

    Unsloth Studio issues keys from its own UI (it has its own accounts), so we can
    only use one the user pasted into the dashboard config — we never mint it here.
    """
    # NOTA: qui c'era load_cfg(), che in questo file non esiste — si chiama
    # load_conf(). Chiunque avesse impostato la chiave avrebbe preso un
    # NameError invece della chiave, quindi la chat non poteva funzionare in
    # nessun caso. CONFIG e' il dizionario gia' in memoria, aggiornato da
    # save_conf(): non serve rileggere il file.
    return (chiave_unsloth() or os.environ.get("UNSLOTH_API_KEY") or "").strip()


def ai_chat(model, messages):
    if not isinstance(messages, list):
        return {"ok": False, "error": "messaggi non validi"}

    if ai_engine() == "unsloth":
        if not _port_open(UNSLOTH_PORT):
            return {"ok": False, "error": "Motore AI spento — accendilo dal modulo AI."}
        key = _unsloth_key()
        if not key:
            return {"ok": False, "error": "Unsloth richiede una API key: creala in Unsloth Studio "
                                          "(pulsante «Apri Unsloth Studio»), poi scrivila in "
                                          "/etc/skillfish/unsloth.key, che resta leggibile solo da root. "
                                          "Non metterla in dashboard.json: quel file lo legge chiunque."}
        try:
            body = {"messages": messages, "stream": False}
            if model:
                body["model"] = model
            req = urllib.request.Request(
                "http://127.0.0.1:%d/v1/chat/completions" % UNSLOTH_PORT,
                data=json.dumps(body).encode(),
                headers={"Content-Type": "application/json", "Authorization": "Bearer " + key})
            out = json.loads(urllib.request.urlopen(req, timeout=300).read().decode())
            msg = (out.get("choices") or [{}])[0].get("message", {})
            # reasoning models put the answer in reasoning_content when they run out of budget
            return {"ok": True, "model": out.get("model", model or ""),
                    "message": msg.get("content") or msg.get("reasoning_content", "")}
        except Exception as e:
            return {"ok": False, "error": "Unsloth: %s" % e}

    # ai_engine() oggi risponde sempre «unsloth», ma e' rimasta una funzione
    # apposta perche' domani il motore puo' cambiare. Il giorno in cui risponde
    # altro, senza questa riga si tornava None: chi chiama fa r.get("ok") e si
    # prende un AttributeError, cioe' un errore 500 al posto di una frase.
    return {"ok": False, "error": "Motore AI sconosciuto: %s" % ai_engine()}


# ---------------- Wake-on-LAN / power schedule ----------------
def _primary_nic():
    out = subprocess.run("ip -o route get 1.1.1.1 2>/dev/null", shell=True, capture_output=True, text=True).stdout
    m = re.search(r"dev (\S+)", out)
    return m.group(1) if m else "enp4s0"


def wol_info():
    nic = _primary_nic()
    try:
        with open("/sys/class/net/%s/address" % nic) as f:
            mac = f.read().strip()
    except Exception:
        mac = ""
    eth = ""
    try:
        eth = subprocess.run(["ethtool", nic], capture_output=True, text=True, timeout=5).stdout
    except Exception:
        eth = ""  # ethtool may be missing; report what we can
    return {"ok": True, "nic": nic, "mac": mac,
            "wol_supported": "Supports Wake-on" in eth,
            "wol_enabled": bool(re.search(r"Wake-on:\s*\w*g", eth))}


def wol_enable(on):
    try:
        subprocess.run(["ethtool", "-s", _primary_nic(), "wol", "g" if on else "d"],
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5)
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": str(e)}


def wol_send(mac):
    if not re.match(r"^([0-9a-fA-F]{2}[:-]){5}[0-9a-fA-F]{2}$", mac or ""):
        return {"ok": False, "error": "MAC non valido"}
    subprocess.run(["wakeonlan", mac], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return {"ok": True, "mac": mac}


def power_schedule(action, minutes):
    try:
        m = max(1, int(minutes))
    except Exception:
        m = 1
    if action == "cancel":
        subprocess.run(["shutdown", "-c"]); return {"ok": True, "cancelled": True}
    if action == "reboot":
        subprocess.run(["shutdown", "-r", "+%d" % m]); return {"ok": True, "in_min": m}
    if action in ("poweroff", "shutdown"):
        subprocess.run(["shutdown", "-h", "+%d" % m]); return {"ok": True, "in_min": m}
    return {"ok": False, "error": "azione sconosciuta"}


# ---------------- auto rules + screen snapshot ----------------
import urllib.request
RULES_DEFAULT = {"enabled": True, "temp_limit": 92, "samples": 6}
_RULES = {"hot": 0, "last_action": ""}


def save_conf():
    u"""Scrive la configurazione. Non solleva, ma non tace piu'.

    Un salvataggio fallito in silenzio si presenta all'utente come un'opzione
    che si rimette da sola al riavvio: il posto peggiore dove cercare la causa.
    """
    try:
        with open(CONF, "w") as f:
            json.dump(CONFIG, f, indent=2)
        # ⚠️ 0644, e non e' una svista. Questo file lo legge anche
        # skillfish-remote-ctl lanciato dall'utente, che e' come
        # l'applicazione sa quali moduli sono accesi. Quando era 0600 root non
        # ci riusciva e disegnava tutte le caselle vuote. Il segreto sta in
        # unsloth.key, che resta 0600.
        try:
            os.chmod(CONF, 0o644)
        except OSError:
            pass
        return True
    except OSError as e:
        sys.stderr.write("dashboard: configurazione non salvata in %s (%s)\n"
                         % (CONF, e))
        return False


# ---------------- chi puo' entrare ----------------
# ⚠️ NON ci si lega a un indirizzo: un socket si lega a UNO solo, e le reti da
# servire sono almeno due (la LAN e la ZeroTier), con la seconda che puo'
# comparire dopo l'avvio. Quindi si ascolta ovunque e si filtra chi bussa.
# ⚠️ NON chiamarlo KEY_F: quel nome e' gia' preso dalla chiave TLS del
# servizio, in cima al file. Chiamandolo cosi' la generazione del certificato
# ha scritto la chiave privata TLS dentro unsloth.key, senza un errore.
UNSLOTH_KEY_F = os.path.join(STATE, "unsloth.key")
_RETI = {"quando": 0.0, "liste": []}


def reti_locali():
    u"""Le reti a cui questa macchina e' attaccata, come stringhe CIDR.

    Sono esattamente quelle che l'utente considera «casa»: la LAN, e la ZeroTier
    se c'e' (e' un'interfaccia come le altre, quindi entra da sola). Il loopback
    lo aggiunge chi chiama.
    """
    fuori = []
    try:
        p = subprocess.run(["ip", "-o", "-f", "inet", "addr", "show"],
                           capture_output=True, text=True, timeout=5)
        for riga in p.stdout.splitlines():
            campi = riga.split()
            if len(campi) < 4 or campi[1] == "lo":
                continue
            try:
                fuori.append(str(ipaddress.ip_interface(campi[3]).network))
            except ValueError:
                continue
    except Exception:
        pass
    return fuori


def reti_ammesse():
    u"""Le reti da cui accettiamo, con la cache: `ip addr` a ogni richiesta
    sarebbe un processo per clic."""
    ora = time.time()
    if ora - _RETI["quando"] < 60 and _RETI["liste"]:
        return _RETI["liste"]
    voci = ["127.0.0.0/8", "::1/128"]
    if CONFIG.get("reti_auto", True):
        voci += reti_locali()
    for r in CONFIG.get("reti_extra", []) or []:
        voci.append(str(r))
    reti = []
    for v in voci:
        try:
            reti.append(ipaddress.ip_network(v, strict=False))
        except ValueError:
            sys.stderr.write("rete non valida, la salto: %s\n" % v)
    _RETI["quando"], _RETI["liste"] = ora, reti
    return reti


def puo_entrare(indirizzo):
    try:
        ip = ipaddress.ip_address(indirizzo)
    except ValueError:
        return False
    return any(ip in r for r in reti_ammesse())


def chiave_unsloth():
    u"""La chiave API sta in un file suo, 0600.

    ⚠️ Stava dentro dashboard.json, e per proteggerla il file era 0600 root. Ma
    quel file lo deve leggere anche l'applicazione dell'utente, per sapere quali
    moduli sono accesi: non potendo, disegnava tutte le caselle vuote. Un
    segreto e una configurazione pubblica non stanno nello stesso file.
    """
    try:
        with open(UNSLOTH_KEY_F) as f:
            return f.read().strip()
    except OSError:
        return ""


def salva_chiave_unsloth(valore):
    try:
        if not valore:
            if os.path.exists(UNSLOTH_KEY_F):
                os.remove(UNSLOTH_KEY_F)
            return True
        fd = os.open(UNSLOTH_KEY_F, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(valore.strip() + "\n")
        return True
    except OSError as e:
        sys.stderr.write("non salvo la chiave: %s\n" % e)
        return False


def migra_chiave():
    u"""Chi aggiorna ha ancora la chiave dentro dashboard.json: si sposta una
    volta sola, e il file torna leggibile."""
    # I moduli nati dopo la configurazione dell'utente: mancando, valevano
    # «spento» per sempre. Si aggiunge solo cio' che MANCA, cosi' chi ha spento
    # qualcosa apposta se lo ritrova spento. E si buttano le chiavi morte.
    mod = CONFIG.setdefault("modules", {})
    for k, v in MODULI_PREDEFINITI.items():
        mod.setdefault(k, v)
    for k in [k for k in mod if k not in MODULE_META]:
        del mod[k]
    vecchia = (CONFIG.pop("unsloth_api_key", "") or "").strip()
    if vecchia:
        salva_chiave_unsloth(vecchia)
        sys.stderr.write("chiave Unsloth spostata in %s\n" % UNSLOTH_KEY_F)
    save_conf()




def rules_cfg():
    return {**RULES_DEFAULT, **(CONFIG.get("rules_cfg") or {})}


def rules_loop():
    while True:
        try:
            c = rules_cfg()
            if CONFIG.get("modules", {}).get("rules") and c["enabled"]:
                v = read_all()
                temps = [x for x in (v.get("gpu_temp"), v.get("cpu_temp")) if isinstance(x, (int, float))]
                t = max(temps) if temps else 0
                if t >= c["temp_limit"]:
                    _RULES["hot"] += 1
                    if _RULES["hot"] >= c["samples"]:
                        apply_preset("Stock")
                        _RULES["hot"] = 0
                        _RULES["last_action"] = "%s — %d°C ≥ %d°C → preset Stock applicato" % (
                            time.strftime("%H:%M:%S"), round(t), c["temp_limit"])
                else:
                    _RULES["hot"] = 0
        except Exception:
            pass
        time.sleep(2)


# ---------------- AI-Ops (local LLM log diagnosis) ----------------
def aiops_diagnose(question):
    # Prima interrogava Ollama sulla 11434 senza autenticazione. Unsloth espone
    # la stessa forma di API sulla 8888 ma vuole una chiave, quindi passiamo da
    # ai_chat(), che la chiave la gestisce gia' e sa dire all'utente dove
    # incollarla se manca.
    if not _port_open(UNSLOTH_PORT):
        return {"ok": False, "error": "Motore AI spento: accendilo dal modulo AI."}
    jrn = subprocess.run(["journalctl", "-p", "warning", "-n", "60", "--no-pager", "-o", "short"],
                         capture_output=True, text=True, timeout=8).stdout
    try:
        with open("/var/log/skillfish-freeze.log") as f:
            frz = f.read()[-1500:]
    except Exception:
        frz = "(nessun freeze registrato)"
    v = read_all()
    tele = "GPU %s°C / CPU %s°C, GPU %s MHz, %s W, ventola %s RPM" % (
        v.get("gpu_temp"), v.get("cpu_temp"), v.get("gpu_freq"), v.get("gpu_power"), v.get("fan"))
    q = question or "Analizza i log e lo stato: ci sono problemi? Causa probabile e come risolvere?"
    prompt = ("Sei l'assistente di sistema di SkillFishOS su una scheda AMD BC-250. "
              "Rispondi in italiano, conciso e pratico.\n\n=== Telemetria ===\n%s\n\n"
              "=== Log freeze ===\n%s\n\n=== journalctl (warning+) ===\n%s\n\n=== Domanda ===\n%s\n" %
              (tele, frz, jrn[-3000:], q))
    r = ai_chat("", [{"role": "user", "content": prompt}])
    if not r.get("ok"):
        return r
    return {"ok": True, "model": r.get("model", ""), "answer": (r.get("message") or "").strip()}


# ---------------- ZeroTier (remote access from anywhere) ----------------
_ZT_CMDS = {"info", "listnetworks", "join", "leave"}


def _zt(args):
    # hardening: only allow known subcommands and strictly-shaped arguments
    if not args or args[0] not in _ZT_CMDS:
        return ""
    for a in args[1:]:
        if not re.match(r"^[0-9a-fA-F]{16}$", a):  # only 16-hex network IDs
            return ""
    try:
        return subprocess.run(["zerotier-cli"] + list(args), capture_output=True,
                              text=True, timeout=8).stdout.strip()
    except Exception:
        return ""


def zt_status():
    info = _zt(["info"])
    parts = info.split()
    address = parts[2] if len(parts) > 2 else ""
    nets = []
    for ln in _zt(["listnetworks"]).splitlines():
        p = ln.split()
        # 200 listnetworks <nwid> <name> <mac> <status> <type> <dev> <ips>
        if len(p) >= 8 and p[1] == "listnetworks" and p[2] != "<nwid>":
            ip = p[8] if len(p) > 8 else "-"
            nets.append({"nwid": p[2], "name": p[3], "status": p[5], "ip": ip})
    return {"ok": True, "address": address, "online": "ONLINE" in info, "networks": nets}


def zt_join(nwid):
    if not re.match(r"^[0-9a-fA-F]{16}$", (nwid or "").strip()):
        return {"ok": False, "error": "Network ID non valido (16 cifre esadecimali)"}
    _zt(["join", nwid.strip()])
    return {"ok": True, "nwid": nwid.strip()}


def zt_leave(nwid):
    if re.match(r"^[0-9a-fA-F]{16}$", (nwid or "").strip()):
        _zt(["leave", nwid.strip()])
    return {"ok": True}


# ---------------- HTTP ----------------
CONFIG = load_conf()
_login_fails = {}  # ip -> (count, first_ts)

# ============================ IL HUD ============================
# Il configuratore del pannello sulla scrivania, servito da qui.
#
# ⚠️ IL HUD E' DELL'UTENTE, NON DI ROOT.
# Le preferenze stanno in ~/.config/skillfish/hud.json e conky gira dentro alla
# sessione grafica. Questo demone gira da root e una sessione grafica non ce
# l'ha: scrivere nella home di root vorrebbe dire salvare in un posto che non
# guarda nessuno, e il pannello non cambierebbe mai. Percio' qui si cerca chi ha
# davvero il desktop, si scrive nella SUA home lasciandogli i file, e conky lo si
# riavvia nel SUO ambiente.
sys.path.insert(0, "/usr/share/skillfish")
try:
    import hud_dati
except ImportError:      # senza il modulo la pagina si spegne, il resto vive
    hud_dati = None

HUD_GEN = "/usr/local/bin/skillfish-hud-config"
HUD_AVVIO = "/usr/local/bin/skillfish-hud"


def hud_utente():
    u"""Chi ha il desktop su questa macchina.

    Tre tentativi, dal piu' attendibile al piu' probabile: la sessione grafica
    secondo systemd, chi fa girare conky adesso, e infine il primo utente vero.
    ⚠️ Nessuno di questi e' root: se il risultato fosse root avremmo scritto in
    una home senza schermo davanti.
    """
    try:
        p = subprocess.run(["loginctl", "list-sessions", "--no-legend"],
                           capture_output=True, text=True, timeout=6)
        for riga in (p.stdout or "").splitlines():
            campi = riga.split()
            if len(campi) >= 3 and campi[2] != "root":
                if len(campi) < 4 or campi[3].startswith("seat"):
                    return campi[2]
    except Exception:
        pass
    try:
        p = subprocess.run(["ps", "-eo", "user,comm"], capture_output=True,
                           text=True, timeout=6)
        for riga in (p.stdout or "").splitlines():
            campi = riga.split()
            if len(campi) == 2 and campi[1] == "conky" and campi[0] != "root":
                return campi[0]
    except Exception:
        pass
    try:
        import pwd as _pwd
        return _pwd.getpwuid(1000).pw_name
    except Exception:
        return ""


def hud_ambiente(utente):
    u"""DISPLAY e bus della sessione, presi da un processo che ci gira dentro.

    ⚠️ Non si inventano: DISPLAY=:0 e' vero quasi sempre e «quasi» qui non basta
    (Wayland, un secondo schermo, un utente entrato due volte). Si legge
    l'ambiente di un processo della sessione, che quei valori li ha per forza
    giusti — e se non c'e' nessun processo, non c'e' nessuna sessione da
    aggiornare.
    """
    fuori = {}
    try:
        p = subprocess.run(["pgrep", "-u", utente, "-x", "plasmashell"],
                           capture_output=True, text=True, timeout=6)
        pid = (p.stdout or "").split()
        if not pid:
            p = subprocess.run(["pgrep", "-u", utente, "-x", "conky"],
                               capture_output=True, text=True, timeout=6)
            pid = (p.stdout or "").split()
        if not pid:
            return fuori
        with open("/proc/%s/environ" % pid[0], "rb") as f:
            for voce in f.read().split(b"\0"):
                if b"=" not in voce:
                    continue
                k, v = voce.decode("utf-8", "replace").split("=", 1)
                if k in ("DISPLAY", "WAYLAND_DISPLAY", "XDG_RUNTIME_DIR",
                         "DBUS_SESSION_BUS_ADDRESS", "XAUTHORITY"):
                    fuori[k] = v
    except Exception:
        pass
    return fuori


def hud_come_utente(utente, argv, ambiente=None, aspetta=True):
    u"""Esegue qualcosa come l'utente del desktop, nel suo ambiente.

    ⚠️ `aspetta=False` per cio' che NON finisce. Il HUD e' conky, e conky
    resta li' finche' non lo si spegne: aspettarlo voleva dire restare
    appesi fino al tempo massimo e poi ucciderlo. Il risultato era un
    pannello che spariva e non tornava piu', dopo aver risposto che era
    andato tutto bene.
    """
    import pwd as _pwd
    try:
        u = _pwd.getpwnam(utente)
    except KeyError:
        return 1, "utente sconosciuto: %s" % utente
    env = {"HOME": u.pw_dir, "USER": utente, "LOGNAME": utente,
           "PATH": "/usr/local/bin:/usr/bin:/bin",
           "LANG": os.environ.get("LANG", "C.UTF-8")}
    env.update(ambiente or {})

    def diventa():
        os.setgid(u.pw_gid)
        try:
            os.initgroups(utente, u.pw_gid)
        except Exception:
            pass
        os.setuid(u.pw_uid)

    if not aspetta:
        try:
            subprocess.Popen(argv, preexec_fn=diventa, env=env, cwd=u.pw_dir,
                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                             stdin=subprocess.DEVNULL, start_new_session=True)
            return 0, ""
        except Exception as e:
            return 1, str(e)
    try:
        p = subprocess.run(argv, preexec_fn=diventa, env=env, cwd=u.pw_dir,
                           capture_output=True, text=True, timeout=90)
        return p.returncode, ((p.stdout or "") + (p.stderr or "")).strip()
    except Exception as e:
        return 1, str(e)


def hud_stato():
    u"""Tutto quello che serve alla pagina: scelte, disponibilita', valori."""
    if hud_dati is None:
        return {"ok": False, "errore": "manca /usr/share/skillfish/hud_dati.py"}
    utente = hud_utente()
    if not utente:
        return {"ok": False, "errore": "nessun utente con il desktop"}
    import pwd as _pwd
    try:
        home = _pwd.getpwnam(utente).pw_dir
    except KeyError:
        return {"ok": False, "errore": "utente sconosciuto: %s" % utente}
    d = hud_dati.dati()
    pref = hud_dati.leggi_pref(hud_dati.percorso_pref(home))
    return {
        "ok": True,
        "utente": utente,
        "voci": hud_dati.CHIAVI,
        "predefinito": hud_dati.ORDINE_PREDEFINITO,
        "posizioni": hud_dati.POSIZIONI,
        "disponibili": hud_dati.disponibili(d),
        "dati": dict((k, v) for k, v in d.items() if not isinstance(v, bool)),
        "gpu_util_c_e": bool(d.get("gpu_util_c_e")),
        "bluetooth_c_e": bool(d.get("bluetooth")),
        "pref": hud_dati.normalizza(pref) if pref else {},
        "acceso": hud_dati.acceso(hud_dati.percorso_autostart(home)),
        "conky": bool(subprocess.run(["pgrep", "-u", utente, "-x", "conky"],
                                     capture_output=True).returncode == 0),
    }


def hud_accendi(utente, home, acceso):
    u"""Scrive l'avvio automatico e fa partire o fermare conky ADESSO.

    ⚠️ Sono due cose separate. Spegnere senza fermarlo lascia il HUD a schermo
    fino al riavvio; fermarlo senza scrivere lo fa tornare da solo domani. Farne
    una sola e' sbagliato in tutti e due i modi.
    """
    auto = hud_dati.percorso_autostart(home)
    skel = "/etc/skel/.config/autostart/skillfish-conky.desktop"
    cartella = os.path.dirname(auto)
    if not os.path.isdir(cartella):
        hud_come_utente(utente, ["mkdir", "-p", cartella])
    if not os.path.exists(auto) and os.path.exists(skel):
        # ⚠️ Si ricopia da /etc/skel invece di riscriverlo: quel file ha dentro
        # una lezione (il percorso assoluto senza virgolette, perche' KDE
        # trasforma l'avvio automatico in un servizio systemd e li' $HOME resta
        # letterale) e riscriverlo a mano la perderebbe.
        hud_come_utente(utente, ["cp", skel, auto])
    try:
        with open(auto, encoding="utf-8") as f:
            righe = [r for r in f.read().split("\n")
                     if not r.lower().startswith("hidden=")]
    except (IOError, OSError):
        righe = []
    if righe:
        if not acceso:
            righe.append("Hidden=true")
        testo = "\n".join(righe).rstrip() + "\n"
        try:
            with open(auto, "w", encoding="utf-8") as f:
                f.write(testo)
            import pwd as _pwd
            u = _pwd.getpwnam(utente)
            os.chown(auto, u.pw_uid, u.pw_gid)
        except (IOError, OSError, KeyError):
            pass
    subprocess.run(["pkill", "-u", utente, "-x", "conky"], capture_output=True)
    if acceso:
        amb = hud_ambiente(utente)
        if amb:
            hud_come_utente(utente, [HUD_AVVIO], amb, aspetta=False)


def hud_scrivi(pref_in):
    u"""Salva le scelte, rigenera la configurazione, riavvia il pannello."""
    if hud_dati is None:
        return {"ok": False, "errore": "manca hud_dati.py"}
    utente = hud_utente()
    if not utente:
        return {"ok": False, "errore": "nessun utente con il desktop"}
    import pwd as _pwd
    try:
        u = _pwd.getpwnam(utente)
    except KeyError:
        return {"ok": False, "errore": "utente sconosciuto: %s" % utente}
    pref = hud_dati.normalizza(pref_in or {})
    percorso = hud_dati.percorso_pref(u.pw_dir)
    cartella = os.path.dirname(percorso)
    try:
        if not os.path.isdir(cartella):
            os.makedirs(cartella)
            os.chown(cartella, u.pw_uid, u.pw_gid)
        tmp = percorso + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            f.write(json.dumps(pref, indent=2, ensure_ascii=False, sort_keys=True))
        os.rename(tmp, percorso)
        # ⚠️ I file restano dell'UTENTE: scritti da root resterebbero di root, e
        # la finestra sulla scrivania — che gira da lui — non potrebbe piu'
        # salvare. Da remoto sembrerebbe tutto a posto e in locale sarebbe rotto.
        os.chown(percorso, u.pw_uid, u.pw_gid)
    except (IOError, OSError) as e:
        return {"ok": False, "errore": str(e)}
    # la configurazione la scrive skillfish-hud-config, che e' l'unico posto che
    # sa quali dati la macchina ha davvero e come diventano righe di conky
    rc, uscita = hud_come_utente(utente, [HUD_GEN])
    if rc != 0:
        return {"ok": False, "errore": uscita[:400] or "skillfish-hud-config ha fallito"}
    hud_accendi(utente, u.pw_dir, pref["mostra"])
    return {"ok": True, "pref": pref, "utente": utente}


class Handler(BaseHTTPRequestHandler):
    def handle_one_request(self):
        # ⚠️ Il filtro sta QUI e non nei singoli percorsi: dimenticarlo in uno
        # solo vorrebbe dire lasciare una porta aperta senza accorgersene.
        chi = self.client_address[0] if self.client_address else ""
        if not puo_entrare(chi):
            sys.stderr.write("rifiutato: %s non e' in nessuna rete ammessa\n" % chi)
            try:
                self.rfile.readline(65537)
                # ⚠️ Niente send_response() qui: si appoggia a requestline,
                # command e request_version, che non abbiamo fatto analizzare.
                # E il buffer va SVUOTATO prima di chiudere, se no il client
                # vede una connessione caduta invece di un rifiuto, e va a
                # cercare il guasto nella rete invece che nelle regole.
                corpo = b"Not on an allowed network.\n"
                self.wfile.write(b"HTTP/1.1 403 Forbidden\r\n"
                                 b"Content-Type: text/plain; charset=utf-8\r\n"
                                 b"Content-Length: " + str(len(corpo)).encode() + b"\r\n"
                                 b"Connection: close\r\n\r\n" + corpo)
                self.wfile.flush()
            except Exception:
                pass
            self.close_connection = True
            return
        # senza «return»: il metodo di base restituisce sempre None e il
        # valore non lo guarda nessuno
        BaseHTTPRequestHandler.handle_one_request(self)

    server_version = "SkillFishRemote/1.0"
    protocol_version = "HTTP/1.1"

    def log_message(self, *a):
        pass

    # --- helpers ---
    def _send(self, code, body=b"", ctype="application/json", extra=None):
        if isinstance(body, str):
            body = body.encode()
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        for k, v in (extra or {}):
            # strip CR/LF to prevent HTTP response splitting / header injection
            self.send_header(str(k).replace("\r", "").replace("\n", ""),
                             str(v).replace("\r", "").replace("\n", ""))
        self.end_headers()
        if body:
            self.wfile.write(body)

    def _json(self, code, obj, extra=None):
        self._send(code, json.dumps(obj), "application/json", extra)

    def _user(self):
        c = self.headers.get("Cookie", "")
        m = re.search(r"sfdash=([^;]+)", c)
        return check_token(m.group(1)) if m else None

    def _body(self):
        n = int(self.headers.get("Content-Length", 0) or 0)
        return self.rfile.read(n) if n else b""

    def _mod(self, name):
        return bool(CONFIG.get("modules", {}).get(name))

    def _guard(self, mod):
        """Return True if request is allowed (authed + module on), else send error."""
        if not self._user():
            self._json(401, {"error": "auth"}); return False
        if mod and not self._mod(mod):
            self._json(403, {"error": "modulo disattivato"}); return False
        return True

    def _proxy(self, port, strip_prefix=None):
        """Reverse-proxy this request to a localhost backend (ttyd/websockify), tunnelling
        WebSocket upgrades as raw TCP. Auth is already enforced by the caller (single sign-on:
        the dashboard session gates these, so the backends need no auth of their own)."""
        path = self.path
        if strip_prefix and path.startswith(strip_prefix):
            path = path[len(strip_prefix):] or "/"
            if not path.startswith("/"):
                path = "/" + path
        is_ws = self.headers.get("Upgrade", "").lower() == "websocket"
        try:
            b = socket.create_connection(("127.0.0.1", port), timeout=10)
        except Exception:
            return self._json(502, {"error": "servizio non avviato"})
        out = ["%s %s HTTP/1.1" % (self.command, path)]
        for k, v in self.headers.items():
            if k.lower() in ("host", "connection"):
                continue
            out.append("%s: %s" % (k, v))
        out.append("Host: 127.0.0.1:%d" % port)
        out.append("Connection: " + ("Upgrade" if is_ws else "close"))
        body = b""
        clen = int(self.headers.get("Content-Length", 0) or 0)
        if clen:
            body = self.rfile.read(clen)
        b.sendall(("\r\n".join(out) + "\r\n\r\n").encode() + body)
        self.close_connection = True
        cli = self.connection
        if is_ws:
            b.setblocking(False); cli.setblocking(False)
            try:
                while True:
                    r, _, x = select.select([cli, b], [], [cli, b], 300)
                    if x or not r:
                        break
                    stop = False
                    for s in r:
                        try:
                            data = s.recv(65536)
                        except Exception:
                            data = b""
                        if not data:
                            stop = True; break
                        (b if s is cli else cli).sendall(data)
                    if stop:
                        break
            except Exception:
                pass
            finally:
                try: b.close()
                except Exception: pass
        else:
            try:
                while True:
                    chunk = b.recv(65536)
                    if not chunk:
                        break
                    cli.sendall(chunk)
            except Exception:
                pass
            finally:
                try: b.close()
                except Exception: pass

    def _client_ip(self):
        return self.client_address[0]

    # --- routing ---
    def do_GET(self):
        path = urllib.parse.urlparse(self.path).path
        # reverse-proxied interactive services (single sign-on via the dashboard session)
        if path == "/terminal" or path.startswith("/terminal/"):
            if not self._guard("terminal"):
                return
            if not _alive("ttyd"):
                terminal_start()
            return self._proxy(TTYD_PORT)
        if path == "/kvm" or path.startswith("/kvm/"):
            if not self._guard("kvm"):
                return
            # Either half may be gone (issue #111): x11vnc exits when it starts
            # before the desktop session exists, while websockify stays up with
            # nothing behind it. _spawn() skips the half that is still alive.
            if not (_alive("x11vnc") and _alive("websockify")):
                kvm_start()
            return self._proxy(KVM_PORT, "/kvm")
        if path == "/unsloth" or path.startswith("/unsloth/"):
            # Unsloth Studio binds to loopback only; the dashboard is the LAN door and
            # has already authenticated the user via PAM before we get here.
            if not self._guard("ai"):
                return
            if not _port_open(UNSLOTH_PORT):
                ai_start()
            return self._proxy(UNSLOTH_PORT, "/unsloth")
        if path in ("/", "/index.html"):
            return self._serve_file("index.html", "text/html; charset=utf-8")
        if path.startswith("/static/"):
            return self._serve_file(path[len("/static/"):], None)
        if path == "/api/i18n":
            # Il dizionario condiviso, per le pagine web. Senza autenticazione
            # come /static/: sono gli stessi file che stanno nei pacchetti.
            lang = urllib.parse.parse_qs(
                urllib.parse.urlparse(self.path).query).get("lang", [""])[0]
            return self._json(200, {"lang": lang, "voci": i18n_voci(lang)})
        if path == "/api/me":
            u = self._user()
            return self._json(200 if u else 401, {"user": u} if u else {"error": "auth"})
        if path == "/api/modules":
            if not self._user():
                return self._json(401, {"error": "auth"})
            mods = [{"id": k, **MODULE_META.get(k, {"name": k, "icon": "•"})}
                    for k, on in CONFIG.get("modules", {}).items()
                    if on and k in MODULE_META]
            return self._json(200, {"modules": mods, "host": sysinfo().get("host", "")})
        if path == "/api/status":
            if not self._user():
                return self._json(401, {"error": "auth"})
            s = sysinfo()
            s["you"] = (self.headers.get("Host", "") or "").split(":")[0]  # host you actually reached us on
            return self._json(200, s)
        if path == "/api/telemetry":
            if not self._guard("telemetry"):
                return
            return self._sse_telemetry()
        if path == "/api/gddr6":
            if not self._guard("telemetry"):
                return
            return self._json(200, gddr6_stato())
        if path == "/api/sicurezza":
            if not self._guard("telemetry"):
                return
            return self._json(200, {"stato": psp_sicurezza()})
        if path == "/api/tuner":
            if not self._guard("tuner"):
                return
            ps = [{"name": p.get("name"), "desc": p.get("desc_it") or p.get("desc", "")} for p in list_presets()]
            return self._json(200, {"presets": ps, "state": tuner_cmd([{"cmd": "get"}])})
        if path == "/api/tuner/cu-profiles":
            if not self._guard("tuner"):
                return
            return self._json(200, {"ok": True, "profiles": cu_profiles_load()})
        if path == "/api/tuner/coremap":
            if not self._guard("tuner"):
                return
            return self._json(200, cpu_coremap())
        if path == "/api/tuner/cpu-cores":
            # GET reads the live core map; the POST route below changes it
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "cpu-cores"}]))
        if path == "/api/tuner/gpu-calibrate":
            if not self._guard("tuner"):
                return
            return self._json(200, calibra_stato())
        if path == "/api/tuner/scx":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "scx"}]))
        if path == "/api/tuner/cpu-gov":
            # the CPU frequency governor (issue #95), same helper as the window
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "cpu-gov"}]))
        if path == "/api/tuner/cpu-prova":
            if not self._guard("tuner"):
                return
            return self._json(200, PROVA_CPU.stato())
        if path == "/api/tuner/coreunlock":
            if not self._guard("tuner"):
                return
            return self._json(200, cc_coreunlock())
        if path == "/api/tuner/gov":
            if not self._guard("tuner"):
                return
            return self._json(200, cc_gov())
        if path == "/api/giochi":
            if not self._guard("giochi"):
                return
            return self._json(200, cc_giochi())
        if path == "/api/profili":
            if not self._guard("profili"):
                return
            return self._json(200, cc_profili())
        if path == "/api/snapshots":
            if not self._guard("snapshots"):
                return
            return self._json(200, snapshots_leggi())
        if path == "/api/hud":
            if not self._guard("hud"):
                return
            return self._json(200, hud_stato())
        if path == "/api/ventola":
            if not self._guard("ventola"):
                return
            s = ventola_stato()
            return self._json(200, {
                "ok": bool(s),
                "demone": ventola_demone(),
                "stato": s,
                "conf": ventola_conf(),
                "prova": ventola_prova_esito(),
            })
        if path == "/api/tuner/fan-curve":
            if not self._guard("tuner"):
                return
            cur = fan_curve_load()
            return self._json(200, {"ok": True, "curve": cur, "preview": fan_curve_pct(cur)})
        if path == "/api/logs":
            if not self._guard("logs"):
                return
            q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
            return self._json(200, {"lines": read_log(q.get("which", ["journal"])[0], q.get("n", ["200"])[0])})
        if path == "/api/ai":
            if not self._guard("ai"):
                return
            return self._json(200, ai_status())
        if path == "/api/ai/update":
            if not self._guard("ai"):
                return
            return self._json(200, ai_update_state())
        if path == "/api/cluster":
            if not self._guard("ai"):
                return
            return self._json(200, cluster_stato())
        if path == "/api/ai/mode":
            if not self._guard("ai"):
                return
            return self._json(200, ai_mode("stato"))
        if path == "/api/ai/conf":
            if not self._guard("ai"):
                return
            return self._json(200, ai_conf())
        if path == "/api/ai/downloads":
            if not self._guard("ai"):
                return
            return self._json(200, ai_downloads())
        if path == "/api/ai/variants":
            if not self._guard("ai"):
                return
            q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
            return self._json(200, ai_variants((q.get("repo_id") or [""])[0].strip()))
        if path == "/api/ai/tune":
            if not self._guard("ai"):
                return
            return self._json(200, ai_tune_status())
        if path == "/api/wol":
            if not self._guard("wol"):
                return
            return self._json(200, wol_info())
        if path == "/api/rules":
            if not self._guard("rules"):
                return
            c = rules_cfg()
            c["last_action"] = _RULES["last_action"]
            return self._json(200, c)
        if path == "/api/zerotier":
            if not self._guard("zerotier"):
                return
            return self._json(200, zt_status())
        if path == "/api/hub/status":
            if not self._guard("hub"):
                return
            return self._json(200, hub_status())
        if path == "/api/hub/catalog":
            if not self._guard("hub"):
                return
            qs = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
            p = {k: v[0] for k, v in qs.items()}
            return self._json(200, hub_catalog(p))
        if path == "/api/hub/categories":
            if not self._guard("hub"):
                return
            return self._json(200, hub_categories())
        if path == "/api/hub/app":
            if not self._guard("hub"):
                return
            key = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query).get("key", [""])[0]
            return self._json(200, hub_app(key))
        if path == "/api/hub/icon":
            if not self._guard("hub"):
                return
            key = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query).get("key", [""])[0]
            ip = hub_icon_path(key)
            if not ip:
                return self._json(404, {"error": "no icon"})
            ext = os.path.splitext(ip)[1].lower()
            ct = {".png": "image/png", ".svg": "image/svg+xml", ".jpg": "image/jpeg",
                  ".jpeg": "image/jpeg", ".xpm": "image/x-xpixmap"}.get(ext, "application/octet-stream")
            try:
                with open(ip, "rb") as f:
                    return self._send(200, f.read(), ct, [("Cache-Control", "max-age=86400")])
            except Exception:
                return self._json(404, {"error": "no icon"})
        if path == "/api/hub/updates":
            if not self._guard("hub"):
                return
            return self._json(200, hub_updates())
        if path == "/api/hub/installed":
            if not self._guard("hub"):
                return
            return self._json(200, hub_installed())
        if path == "/api/hub/search":
            if not self._guard("hub"):
                return
            q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query).get("q", [""])[0]
            return self._json(200, hub_search(q))
        if path == "/api/hub/sources":
            if not self._guard("hub"):
                return
            return self._json(200, hub_sources())
        if path == "/api/hub/log":
            if not self._guard("hub"):
                return
            return self._json(200, {"running": _HUBJOB["running"], "title": _HUBJOB["title"],
                                    "log": "\n".join(_HUBJOB["log"]), "done": _HUBJOB["done"],
                                    "rc": _HUBJOB["rc"]})
        if path == "/api/config":
            if not self._user():
                return self._json(401, {"error": "auth"})
            cat = [dict(id=k, **v) for k, v in MODULE_META.items()]
            return self._json(200, {"modules": CONFIG.get("modules", {}), "catalogue": cat})
        return self._json(404, {"error": "not found"})

    def do_POST(self):
        path = urllib.parse.urlparse(self.path).path
        if path == "/api/login":
            return self._login()
        if path == "/api/logout":
            return self._json(200, {"ok": True},
                              extra=[("Set-Cookie", "sfdash=; Max-Age=0; Path=/; HttpOnly; SameSite=Strict")])
        if not self._user():
            return self._json(401, {"error": "auth"})
        try:
            data = json.loads(self._body() or b"{}")
        except Exception:
            data = {}
        if path == "/api/gddr6":
            if not self._guard("telemetry"):
                return
            azione = data.get("azione")
            if azione not in ("avvia", "ferma"):
                return self._json(400, {"ok": False, "errore": "azione: avvia | ferma"})
            return self._json(200, gddr6_helper(azione))
        if path == "/api/tuner/preset":
            if not self._guard("tuner"):
                return
            return self._json(200, apply_preset(data.get("name")))
        if path == "/api/tuner/govmode":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "gov-mode", "mode": data.get("mode", "balanced")}]))
        if path == "/api/tuner/cpu-cores":
            # live CPU hotplug — same idea as the CU grid, but for the CPU side
            if not self._guard("tuner"):
                return
            if "smt" in data:
                return self._json(200, tuner_cmd([{"cmd": "cpu-smt", "on": bool(data["smt"])}]))
            if "cores" in data:
                return self._json(200, tuner_cmd([{"cmd": "cpu-cores-set", "cores": data["cores"]}]))
            return self._json(200, tuner_cmd([{"cmd": "cpu-cores"}]))
        if path == "/api/tuner/gpu-calibrate":
            if not self._guard("tuner"):
                return
            if data.get("azione") == "ferma":
                return self._json(200, calibra_ferma())
            return self._json(200, calibra_avvia())
        if path == "/api/tuner/scx":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "scx-set", "on": bool(data.get("on"))}]))
        if path == "/api/tuner/cpu-gov":
            if not self._guard("tuner"):
                return
            if data.get("pstates"):
                # update-grub runs inside: give it time
                return self._json(200, tuner_cmd([{"cmd": "cpu-pstates-set", "on": True}], timeout=240))
            return self._json(200, tuner_cmd([{"cmd": "cpu-gov-set", "gov": str(data.get("gov", ""))}], timeout=90))
        if path == "/api/hud/conf":
            if not self._guard("hud"):
                return
            return self._json(200, hud_scrivi(data.get("pref") or data))
        if path == "/api/hud/mostra":
            if not self._guard("hud"):
                return
            # ⚠️ Il solo interruttore non tocca le scelte: chi spegne il HUD
            # vuole ritrovare le sue voci quando lo riaccende, non un HUD
            # vuoto o quello di serie.
            s = hud_stato()
            if not s.get("ok"):
                return self._json(200, s)
            pref = s.get("pref") or {}
            pref["mostra"] = bool(data.get("mostra", True))
            if not pref.get("voci"):
                pref["voci"] = list(hud_dati.ORDINE_PREDEFINITO)
            return self._json(200, hud_scrivi(pref))
        if path == "/api/snapshots":
            if not self._guard("snapshots"):
                return
            return self._json(200, snapshots_azione(
                data.get("azione"), data.get("num"), data.get("nota")))
        if path == "/api/ventola/conf":
            if not self._guard("ventola"):
                return
            c = ventola_conf()
            # ⚠️ Si aggiorna solo cio' che e' arrivato: il web manda un pezzo
            # per volta (la curva, oppure il minimo) e non deve poter azzerare
            # i campi che non ha nominato.
            for chiave in ("attivo", "pwm", "sorgente", "curva", "minimo",
                           "isteresi", "emergenza", "preset", "predittivo"):
                if chiave in data:
                    c[chiave] = data[chiave]
            return self._json(200, ventola_scrivi(c))
        if path == "/api/ventola/etichette":
            if not self._guard("ventola"):
                return
            return self._json(200, ventola_etichette(data))
        if path == "/api/ventola/prova":
            if not self._guard("ventola"):
                return
            return self._json(200, ventola_prova(data.get("pwm"), data.get("fan")))
        if path == "/api/tuner/fan":
            if not self._guard("tuner"):
                return
            if ventola_demone():
                return self._json(200, ventola_percentuale(int(data.get("pct", 45)),
                                                           data.get("mode", "auto")))
            return self._json(200, tuner_cmd([{"cmd": "apply-fan", "mode": data.get("mode", "auto"),
                                               "pct": int(data.get("pct", 45))}]))
        if path == "/api/tuner/cpu":
            if not self._guard("tuner"):
                return
            cmd = "persist-cpu" if data.get("persist") else "apply-cpu"
            return self._json(200, tuner_cmd([{"cmd": cmd, "mhz": int(data.get("mhz", 3700)),
                                               "scale": int(data.get("scale", 0)),
                                               "temp": int(data.get("temp", 85))}]))
        if path == "/api/tuner/gpu":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "apply-gpu",
                                               "minmhz": int(data.get("minmhz", 350)),
                                               "minmv": int(data.get("minmv", 700)),
                                               "maxmhz": int(data.get("maxmhz", 2200)),
                                               "maxmv": int(data.get("maxmv", 1000))}]))
        if path == "/api/tuner/cu":
            if not self._guard("tuner"):
                return
            rows = data.get("rows", [])
            try:
                rows = [int(x) & 0x1f for x in rows][:4]
            except Exception:
                rows = []
            return self._json(200, tuner_cmd([{"cmd": "cu-apply", "rows": rows}]))
        if path == "/api/tuner/cu-test":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "cu-test"}], timeout=240))
        if path == "/api/tuner/cu-profile":
            if not self._guard("tuner"):
                return
            return self._json(200, cu_profile_op(data.get("action"), data.get("name"), data.get("rows")))
        if path == "/api/tuner/fan-curve":
            if not self._guard("tuner"):
                return
            return self._json(200, fan_curve_save(data))
        if path == "/api/tuner/vram":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "set-vram", "mb": int(data.get("mb", 8192))}]))
        if path == "/api/tuner/cpu-prova":
            if not self._guard("tuner"):
                return
            return self._json(200, cc_prova_cpu_post(data))
        if path == "/api/tuner/coreunlock":
            if not self._guard("tuner"):
                return
            return self._json(200, cc_coreunlock_post(data))
        if path == "/api/tuner/test-cpu":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "test-cpu", "mhz": int(data.get("mhz", 3700)),
                                               "scale": int(data.get("scale", 0)),
                                               "temp": int(data.get("temp", 85))}], timeout=150))
        if path == "/api/tuner/test-gpu":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "test-gpu",
                                               "minmhz": int(data.get("minmhz", 350)),
                                               "minmv": int(data.get("minmv", 700)),
                                               "maxmhz": int(data.get("maxmhz", 2200)),
                                               "maxmv": int(data.get("maxmv", 1000))}], timeout=150))
        if path == "/api/tuner/suggest-uv":
            if not self._guard("tuner"):
                return
            return self._json(200, tuner_cmd([{"cmd": "suggest-uv", "mhz": int(data.get("mhz", 3700))}], timeout=150))
        if path == "/api/tuner/gov":
            if not self._guard("tuner"):
                return
            return self._json(200, cc_gov_post(data))
        if path == "/api/giochi/mesa":
            if not self._guard("giochi"):
                return
            return self._json(200, cc_mesa_set(data.get("app"), bool(data.get("on"))))
        if path == "/api/giochi/mesa-sistema":
            if not self._guard("giochi"):
                return
            return self._json(200, tuner_cmd([{"cmd": "mesa-sistema-set", "on": bool(data.get("on"))}], timeout=60))
        if path == "/api/giochi/scx-azzera":
            if not self._guard("giochi"):
                return
            return self._json(200, tuner_cmd([{"cmd": "scx-azzera"}]))
        if path == "/api/giochi/fsr4":
            if not self._guard("giochi"):
                return
            return self._json(200, cc_fsr4_set(data.get("app"), bool(data.get("on"))))
        if path == "/api/giochi/proton-default":
            if not self._guard("giochi"):
                return
            return self._json(200, cc_proton_default(data.get("dove"), data.get("nome")))
        if path == "/api/profili":
            if not self._guard("profili"):
                return
            return self._json(200, cc_profilo_applica(data.get("chiave")))
        if path == "/api/hub/op":
            if not self._guard("hub"):
                return
            return self._json(200, hub_op(data.get("op"), data.get("backend", "apt"), data.get("pkg")))
        if path == "/api/hub/refresh":
            if not self._guard("hub"):
                return
            cat_build_async()
            return self._json(200, {"ok": True, "building": True})
        if path == "/api/hub/source":
            if not self._guard("hub"):
                return
            return self._json(200, hub_source_toggle(data.get("name"), bool(data.get("enable"))))
        if path == "/api/power":
            if not self._user():
                return self._json(401, {"error": "auth"})
            a = data.get("action")
            if a == "reboot":
                subprocess.Popen(["systemctl", "reboot"]); return self._json(200, {"ok": True})
            if a in ("poweroff", "shutdown"):
                subprocess.Popen(["systemctl", "poweroff"]); return self._json(200, {"ok": True})
            return self._json(400, {"error": "azione sconosciuta"})
        if path == "/api/launch":
            if not self._guard("launcher"):
                return
            apps = {"console": ["/usr/local/bin/skillfish-gaming-mode"],
                    "monitor": ["/usr/local/bin/skillfish-monitor"],
                    "tuner": ["/usr/local/bin/skillfish-tuner"],
                    "hub": ["/usr/local/bin/skillfish-hub"],
                    "ai": ["/usr/local/bin/skillfish-ai-panel"]}
            cmd = apps.get(data.get("what"))
            return self._json(200, launch_app(cmd) if cmd else {"ok": False, "error": "app sconosciuta"})
        if path == "/api/kvm/start":
            if not self._guard("kvm"):
                return
            return self._json(200, kvm_start())
        if path == "/api/kvm/stop":
            if not self._guard("kvm"):
                return
            return self._json(200, kvm_stop())
        if path == "/api/terminal/start":
            if not self._guard("terminal"):
                return
            return self._json(200, terminal_start())
        if path == "/api/terminal/stop":
            if not self._guard("terminal"):
                return
            return self._json(200, terminal_stop())
        if path == "/api/ai/start":
            if not self._guard("ai"):
                return
            return self._json(200, ai_start())
        if path == "/api/ai/stop":
            if not self._guard("ai"):
                return
            return self._json(200, ai_stop())
        if path == "/api/ai/tune":
            if not self._guard("ai"):
                return
            return self._json(200, ai_tune_apply(data.get("action")))
        if path == "/api/wol/enable":
            if not self._guard("wol"):
                return
            return self._json(200, wol_enable(bool(data.get("on", True))))
        if path == "/api/wol/send":
            if not self._guard("wol"):
                return
            return self._json(200, wol_send(data.get("mac")))
        if path == "/api/wol/schedule":
            if not self._guard("wol"):
                return
            return self._json(200, power_schedule(data.get("action"), data.get("minutes", 1)))
        if path == "/api/rules":
            if not self._guard("rules"):
                return
            cfg = rules_cfg()
            if "enabled" in data:
                cfg["enabled"] = bool(data["enabled"])
            if "temp_limit" in data:
                cfg["temp_limit"] = max(70, min(100, int(data["temp_limit"])))
            CONFIG["rules_cfg"] = cfg
            save_conf()
            return self._json(200, {"ok": True, **cfg})
        if path == "/api/aiops/diagnose":
            if not self._guard("aiops"):
                return
            return self._json(200, aiops_diagnose(data.get("question")))
        if path == "/api/zerotier/join":
            if not self._guard("zerotier"):
                return
            return self._json(200, zt_join(data.get("nwid")))
        if path == "/api/zerotier/leave":
            if not self._guard("zerotier"):
                return
            return self._json(200, zt_leave(data.get("nwid")))
        if path == "/api/config":
            if not self._user():
                return self._json(401, {"error": "auth"})
            mod = data.get("module")
            if mod in MODULE_META:
                CONFIG.setdefault("modules", {})[mod] = bool(data.get("on"))
                save_conf()
                return self._json(200, {"ok": True, "modules": CONFIG["modules"]})
            return self._json(400, {"error": "modulo sconosciuto"})
        if path == "/api/ai/pull":
            if not self._guard("ai"):
                return
            return self._json(200, ai_pull(data.get("model")))
        if path == "/api/ai/key":
            # Salvare la chiave API di Unsloth dalla dashboard invece che a mano
            # in /etc/skillfish/dashboard.json via SSH. Senza chiave, Chat e
            # AI-Ops prendono 401 da Unsloth e all'utente sembrano due pulsanti
            # rotti: il messaggio d'errore diceva gia' cosa fare, ma nessuno
            # arrivava a leggerlo.
            if not self._guard("ai"):
                return
            k = (data.get("key") or "").strip()
            if k and not re.match(r"^sk-[A-Za-z0-9._-]{8,200}$", k):
                return self._json(400, {"ok": False, "error": "chiave non valida"})
            if k:
                salva_chiave_unsloth(k)
            else:
                salva_chiave_unsloth("")
            save_conf()
            return self._json(200, {"ok": True, "has_key": bool(k)})
        if path == "/api/ai/chat":
            if not self._guard("ai"):
                return
            return self._json(200, ai_chat(data.get("model"), data.get("messages")))
        if path == "/api/ai/update":
            if not self._guard("ai"):
                return
            return self._json(200, ai_update_start())
        if path == "/api/ai/livello":
            if not self._guard("ai"):
                return
            return self._json(200, ai_livello(data.get("livello", ""),
                                              data.get("modello", "")))
        if path == "/api/ai/mode":
            if not self._guard("ai"):
                return
            # ⚠️ E' QUI CHE SI TORNA AL DESKTOP. Con la modalita' accesa il
            # Control Center non esiste piu': questa e' l'unica strada che non
            # passa da ssh, e il demone della dashboard non si ferma mai
            # proprio per questo.
            return self._json(200, ai_mode(data.get("azione", "stato")))
        if path == "/api/ai/conf":
            if not self._guard("ai"):
                return
            return self._json(200, ai_conf_set(data))
        if path == "/api/ai/setup":
            if not self._guard("ai"):
                return
            return self._json(200, ai_setup(data.get("password") or ""))
        if path == "/api/ai/keycreate":
            if not self._guard("ai"):
                return
            return self._json(200, ai_key_create(data.get("password") or ""))
        if path == "/api/ai/download":
            if not self._guard("ai"):
                return
            return self._json(200, ai_download((data.get("repo_id") or "").strip(), data.get("variant")))
        return self._json(404, {"error": "not found"})

    def _login(self):
        ip = self._client_ip()
        cnt, t0 = _login_fails.get(ip, (0, time.time()))
        if cnt >= 8 and time.time() - t0 < 300:
            return self._json(429, {"error": "too many attempts, wait a few minutes"})
        try:
            data = json.loads(self._body() or b"{}")
        except Exception:
            data = {}
        user = (data.get("user") or desktop_user() or "").strip()
        pw = data.get("pass") or ""
        if user and pw and pam_check(user, pw):
            _login_fails.pop(ip, None)
            tok = make_token(user)
            return self._json(200, {"ok": True, "user": user},
                              extra=[("Set-Cookie",
                                      "sfdash=%s; Max-Age=%d; Path=/; HttpOnly; SameSite=Strict; Secure"
                                      % (tok, SESSION_TTL))])
        _login_fails[ip] = (cnt + 1, t0 if cnt else time.time())
        return self._json(401, {"error": "invalid credentials"})

    def _serve_file(self, rel, ctype):
        rel = rel.lstrip("/")
        # canonicalise and ensure the resolved path stays inside WEB (no traversal)
        root = os.path.realpath(WEB)
        full = os.path.realpath(os.path.join(root, rel))
        if full != root and not full.startswith(root + os.sep):
            return self._json(403, {"error": "no"})
        if not os.path.isfile(full):
            return self._json(404, {"error": "not found"})
        if ctype is None:
            ext = os.path.splitext(full)[1]
            ctype = {".js": "application/javascript", ".css": "text/css",
                     ".html": "text/html; charset=utf-8", ".svg": "image/svg+xml",
                     ".png": "image/png", ".ico": "image/x-icon"}.get(ext, "application/octet-stream")
        with open(full, "rb") as f:
            self._send(200, f.read(), ctype)

    def _sse_telemetry(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Connection", "keep-alive")
        self.end_headers()
        st = [None]
        try:
            while True:
                vals = read_all()
                vals["cpu_load"] = cpu_load(st)
                vals["cpu_threads"] = cpu_threads()
                vals["_t"] = round(time.time(), 1)
                self.wfile.write(("data: %s\n\n" % json.dumps(vals)).encode())
                self.wfile.flush()
                time.sleep(0.5)
        except Exception:
            return


def ensure_cert():
    if os.path.isfile(CERT_F) and os.path.isfile(KEY_F):
        return
    host = subprocess.run("hostname", capture_output=True, text=True).stdout.strip() or "skillfishos"
    subprocess.run(
        ["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
         "-keyout", KEY_F, "-out", CERT_F, "-days", "3650",
         "-subj", "/CN=%s" % host,
         "-addext", "subjectAltName=DNS:%s,DNS:%s.local,DNS:localhost" % (host, host)],
        check=False)
    # Questa e' la chiave privata del certificato con cui la dashboard si
    # presenta. Se il permesso non si riesce a stringere, il difetto non e' che
    # il chmod fallisce: e' che la chiave resta leggibile e nessuno lo sa.
    try:
        os.chmod(KEY_F, 0o600)
    except OSError as e:
        sys.stderr.write("dashboard: non riesco a proteggere %s (%s)\n" % (KEY_F, e))
    try:
        modo = os.stat(KEY_F).st_mode & 0o077
        if modo:
            sys.stderr.write("dashboard: ATTENZIONE, la chiave privata %s e' "
                             "leggibile da altri utenti (permessi %o)\n"
                             % (KEY_F, os.stat(KEY_F).st_mode & 0o777))
    except OSError:
        pass


def main():
    os.makedirs(STATE, exist_ok=True)
    ensure_cert()
    migra_chiave()
    bind = CONFIG.get("bind", "0.0.0.0"); port = int(CONFIG.get("port", 8443))
    sys.stderr.write("reti ammesse: %s\n" % ", ".join(str(r) for r in reti_ammesse()))
    httpd = ThreadingHTTPServer((bind, port), Handler)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.minimum_version = ssl.TLSVersion.TLSv1_2  # refuse legacy SSL/TLS
    ctx.load_cert_chain(CERT_F, KEY_F)
    httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
    sys.stderr.write("SkillFish Remote on https://%s:%d (PAM=%s)\n" % (bind, port, _PAM_OK))
    sys.stderr.flush()
    threading.Thread(target=rules_loop, daemon=True).start()
    threading.Thread(target=fan_curve_loop, daemon=True).start()
    if CONFIG.get("modules", {}).get("hub"):
        cat_build_async()  # pre-build the app catalogue in the background
    httpd.serve_forever()


if __name__ == "__main__":
    main()
