#!/usr/bin/env python3
"""SkillFishOS Control Center - the privileged helper (root via pkexec).

One JSON command per line on stdin, one JSON reply per line on stdout, so the
user authenticates ONCE when the first privileged thing is asked and not per
action. It is a superset of the old skillfish-tuner-helper: every command that
one accepted still works, which is what keeps the Remote Manager going.

WHAT IS NEW AGAINST THE TUNER HELPER
- the GPU is the V/F governor (skillfish-vf-governor): a curve of MHz/mV knots
  and a ceiling in /etc/skillfish-vf-governor.json, not two safe-points in a
  TOML. gov-* below read it, validate it, and apply it with a TRIAL: while a
  candidate runs, the file on disk keeps the last known good curve, so a hang
  followed by a power cycle boots into the curve that worked.
- the scheduler is scx_bpfland armed through GameMode (skillfish-scx.path),
  not scx_lavd running all the time.
- our Mesa is the driver of the whole machine on a BC-250 (mesa-sistema), which
  skillfish-mesa switches for the host and the two launchers at once; the
  32-bit side keeps Debian's on purpose.
- kernel, snapshot and fan helpers are reached through here, so one password
  covers the whole window.
"""
import glob
import json
import os
import pwd
import re
import subprocess
import sys
import time

NL = chr(10)

# ---- paths -----------------------------------------------------------------
OC_DIR = next((p for p in ("/opt/bc250_smu_oc", "/root/bc250_smu_oc") if os.path.isdir(p)),
              "/opt/bc250_smu_oc")
OC_CONF = "/etc/bc250-smu-oc.conf"
GOV_CONF = "/etc/skillfish-vf-governor.json"
GOV_BUONO = "/etc/skillfish-vf-governor.json.buono"
GOV_PROVA = "/run/skillfish/gov-prova.json"
GOV_UNIT = "skillfish-vf-governor.service"
GOV_STOCK_UNIT = "cyan-skillfish-governor.service"
GOV_BATTITO = "/run/skillfish-vf-governor.battito"
MEMCFG = next((p for p in ("/usr/local/bin/skillfish-memcfg", "/opt/bc250_memcfg/bc250memcfg")
               if os.path.exists(p)), "/usr/local/bin/skillfish-memcfg")
def find_vkpeak():
    """Where vkpeak is, looked up on every call.

    skillfish-vkpeak installs it in /usr/lib/skillfish/vkpeak (issue #91). The
    two bench folders are where the development boards and the images cloned
    from them had it; no package ever put it there. Looked up each time rather
    than once at start, so installing the package while this helper is running
    is enough.
    """
    for pattern in ("/usr/lib/skillfish/vkpeak/vkpeak", "/opt/bench/vkpeak*/vkpeak", "/root/bench/vkpeak*/vkpeak"):
        found = sorted(glob.glob(pattern))
        if found and os.access(found[0], os.X_OK):
            return found[0]
    return None
SEGNAPOSTO_CORE = "/etc/skillfish/core-unlock.abilitato"
SCX_PATH_UNIT = "skillfish-scx.path"
SCX_SERVICE = "skillfish-scx.service"
SCX_BIN = "/usr/lib/skillfish-scx/scx_bpfland"
SCX_STATO = "/sys/kernel/sched_ext/state"
SCX_OPS = "/sys/kernel/sched_ext/root/ops"
SCX_CONTA = "/var/lib/skillfish/scx-espulsioni"
MESA_NOSTRA = "/opt/skillfish-gfx1013/lib/x86_64-linux-gnu/libvulkan_radeon.so"
MESA_CLI = "/usr/bin/skillfish-mesa"
MESA_PKG = "skillfish-mesa-gfx1013"
KERNEL_HELPER = "/usr/local/bin/skillfish-kernel-helper"
SNAP_HELPER = "/usr/local/bin/skillfish-snapshots-helper"
SNAP_MANUT = "/usr/local/bin/skillfish-btrfs-manutenzione"
FAN_HELPER = "/usr/local/bin/skillfish-fan-helper"
FAN_CONF = "/etc/skillfish/ventola.json"

MV_MIN, MV_MAX = 700, 1129
MHZ_MIN, MHZ_MAX = 350, 2300


# ---- small utilities --------------------------------------------------------
def _rd(path):
    with open(path) as f:
        return f.read()


def _wr(path, data):
    """Write a whole file.

    Configuration files go through a .tmp and a rename, so a crash never leaves
    half a file. Kernel attributes in /sys and /proc are written in place: no
    file can be created there, and the .tmp route failed with EACCES on every
    one of them (issue #93: cpuN/online, and with it smt/control and the CPU
    governor the benchmark sets).
    """
    if path.startswith(("/sys/", "/proc/")):
        with open(path, "w") as f:
            f.write(data)
        return
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        f.write(data)
    os.replace(tmp, path)


def sh(cmd, t=120):
    try:
        return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=t)
    except Exception as e:
        class R:
            returncode = 1
            stdout = ""
            stderr = str(e)
        return R()


def _log(msg):
    """journalctl -t skillfish-cc-helper: everything that touches the hardware."""
    try:
        import syslog
        syslog.openlog("skillfish-cc-helper")
        syslog.syslog(msg)
        syslog.closelog()
    except Exception:
        # logging must never be the reason a privileged command fails
        pass


def _unit(stato, unit):
    return sh("systemctl %s %s" % (stato, unit), 30).returncode == 0


def temp(name):
    for h in glob.glob("/sys/class/hwmon/hwmon*"):
        try:
            if _rd(h + "/name").strip() == name:
                return int(_rd(h + "/temp1_input")) // 1000
        except (OSError, ValueError):
            # this hwmon node is not the one, or not readable: try the next one
            pass
    return 0


def cpu_min_freq():
    fs = [float(x) for x in re.findall(r'cpu MHz\s*:\s*([\d.]+)', _rd("/proc/cpuinfo"))]
    return int(min(fs)) if fs else 0


# ---- the governor -------------------------------------------------------------
def gov_leggi(percorso=GOV_CONF):
    try:
        return json.loads(_rd(percorso))
    except Exception:
        return {}


def gov_valida(c):
    """The curve and the ceiling, checked against the limits of the chip.

    Returns (clean_config, error). Nothing here trusts the caller: the window
    is ours, but the Remote Manager talks to this too, over the network.
    """
    if not isinstance(c, dict):
        return None, "configurazione non valida"
    base = gov_leggi()
    out = dict(base)
    curva = c.get("curva", base.get("curva"))
    try:
        pts = sorted((int(p[0]), int(p[1])) for p in curva)
    except Exception:
        return None, "curva non valida"
    if len(pts) < 2:
        return None, "servono almeno due punti"
    mhz_prec, mv_prec = None, None
    for mhz, mv in pts:
        if not (MHZ_MIN <= mhz <= MHZ_MAX):
            return None, "%d MHz fuori dai limiti %d-%d" % (mhz, MHZ_MIN, MHZ_MAX)
        if not (MV_MIN <= mv <= MV_MAX):
            return None, "%d mV fuori dai limiti %d-%d" % (mv, MV_MIN, MV_MAX)
        if mhz_prec is not None and mhz == mhz_prec:
            return None, "due punti a %d MHz" % mhz
        if mv_prec is not None and mv < mv_prec:
            return None, "la tensione scende salendo di frequenza (%d MHz)" % mhz
        mhz_prec, mv_prec = mhz, mv
    out["curva"] = [[m, v] for m, v in pts]
    try:
        fmax = int(c.get("freq_max", base.get("freq_max", pts[-1][0])))
        fmin = int(c.get("freq_min", base.get("freq_min", pts[0][0])))
    except Exception:
        return None, "tetto non valido"
    if not (pts[0][0] <= fmax <= pts[-1][0]):
        return None, "il tetto %d deve stare fra %d e %d MHz" % (fmax, pts[0][0], pts[-1][0])
    if not (MHZ_MIN <= fmin <= fmax):
        return None, "frequenza minima non valida"
    out["freq_max"], out["freq_min"] = fmax, fmin
    # the tunables: numbers and booleans only, within sane ranges
    numerici = {"gradi_ok": (60, 95), "gradi_max": (65, 97), "gradi_rottura": (70, 99),
                "watt_max": (60, 220), "watt_ok": (40, 200), "margine_salita": (0, 80),
                "passo_su": (50, 2000), "gradino": (10, 300), "conferme_giu": (1, 20),
                "carico_salto": (10, 100), "carico_passo_su": (5, 100),
                "carico_passo_giu": (0, 95), "carico_fondo": (0, 50),
                "droop_max": (0, 200), "droop_su": (0, 50), "droop_giu": (0, 50),
                "droop_da": (350, 2300), "droop_tolleranza": (0, 50)}
    for k, (lo, hi) in numerici.items():
        if k in c:
            try:
                v = int(c[k])
            except Exception:
                return None, "%s non valido" % k
            if not (lo <= v <= hi):
                return None, "%s fuori dai limiti %d-%d" % (k, lo, hi)
            out[k] = v
    for k in ("droop_attivo", "isteresi_fine", "target_uso"):
        if k in c:
            out[k] = bool(c[k])
    if out.get("gradi_ok", 0) >= out.get("gradi_max", 999) or out.get("gradi_max", 0) >= out.get("gradi_rottura", 999):
        return None, "le soglie di temperatura devono crescere: ok < max < rottura"
    return out, None


def gov_scrivi(c, percorso=GOV_CONF):
    _wr(percorso, json.dumps(c, indent=1) + NL)


def gov_riavvia():
    """Stop, let the SMU settle, start. The BC-250 can hang on an abrupt jump.

    ⚠️ reset-failed first. The unit allows three starts in five minutes
    (StartLimitBurst): a trial, its confirmation and one more change are three,
    and the fourth start was refused -- measured 11/09/2026, the governor stayed
    down with the board parked at 350 MHz. reset-failed clears that counter.
    """
    sh("systemctl stop " + GOV_UNIT, 60)
    time.sleep(1.5)
    sh("systemctl reset-failed " + GOV_UNIT, 30)
    return sh("systemctl start " + GOV_UNIT, 60).returncode == 0


def gov_battito():
    try:
        p = _rd(GOV_BATTITO).split()
        return {"quando": float(p[0]), "mhz": int(p[1]), "tetto": int(p[2]),
                "gradi": int(p[3]), "watt": int(p[4]), "carico": int(p[5])}
    except Exception:
        return None


def gov_get():
    prova = None
    try:
        prova = json.loads(_rd(GOV_PROVA))
    except (OSError, json.JSONDecodeError):
        # no trial in progress, or the file is mid-write: report none
        prova = None
    return {"ok": True, "conf": gov_leggi(), "attivo": _unit("is-active", GOV_UNIT),
            "abilitato": _unit("is-enabled", GOV_UNIT),
            "stock_attivo": _unit("is-active", GOV_STOCK_UNIT),
            "installato": os.path.exists("/usr/bin/skillfish-vf-governor"),
            "battito": gov_battito(), "prova": prova,
            "limiti": {"mv": [MV_MIN, MV_MAX], "mhz": [MHZ_MIN, MHZ_MAX]}}


def gov_set(c):
    pulita, err = gov_valida(c)
    if err:
        return {"ok": False, "err": err}
    if os.path.exists(GOV_PROVA):
        return {"ok": False, "err": "c'e' una prova in corso: confermala o annullala prima"}
    _log("curva GPU riscritta: tetto %d, %d punti, ultimo %s" % (
        pulita["freq_max"], len(pulita["curva"]), pulita["curva"][-1]))
    gov_scrivi(pulita)
    return {"ok": gov_riavvia(), "conf": pulita}


def gov_prova(c):
    """Run a candidate curve with the last known good one still on disk."""
    pulita, err = gov_valida(c)
    if err:
        return {"ok": False, "err": err}
    if os.path.exists(GOV_PROVA):
        return {"ok": False, "err": "c'e' gia' una prova in corso"}
    buona = gov_leggi()
    gov_scrivi(buona, GOV_BUONO)
    _log("PROVA curva GPU: tetto %d, ultimo punto %s" % (pulita["freq_max"], pulita["curva"][-1]))
    gov_scrivi(pulita)
    ok = gov_riavvia()
    # the governor has read the candidate: put the good one back on disk NOW,
    # so a hang in the next minute reboots into it
    gov_scrivi(buona)
    os.makedirs(os.path.dirname(GOV_PROVA), exist_ok=True)
    _wr(GOV_PROVA, json.dumps({"candidata": pulita, "da": time.time()}))
    return {"ok": ok, "conf": pulita}


def gov_conferma():
    try:
        cand = json.loads(_rd(GOV_PROVA))["candidata"]
    except Exception:
        return {"ok": False, "err": "nessuna prova in corso"}
    gov_scrivi(cand)
    os.unlink(GOV_PROVA)
    _log("curva GPU confermata: tetto %d" % cand["freq_max"])
    return {"ok": True, "conf": cand}


def gov_annulla():
    if os.path.exists(GOV_PROVA):
        os.unlink(GOV_PROVA)
    _log("prova curva GPU annullata: torno alla curva su disco")
    return {"ok": gov_riavvia(), "conf": gov_leggi()}


def gov_attiva(on):
    if on:
        ok = sh("systemctl enable --now " + GOV_UNIT, 60).returncode == 0
    else:
        ok = sh("systemctl disable --now " + GOV_UNIT, 60).returncode == 0
    return {"ok": ok, **{k: v for k, v in gov_get().items() if k != "ok"}}


def _gpu_compat():
    """The old {min_mhz,min_mv,max_mhz,max_mv} view, for whoever still asks."""
    c = gov_leggi()
    pts = c.get("curva") or [[350, 700], [2100, 1050]]
    fmax = int(c.get("freq_max", pts[-1][0]))
    mv = pts[-1][1]
    for m, v in pts:
        if m <= fmax:
            mv = v
    return {"min_mhz": int(pts[0][0]), "min_mv": int(pts[0][1]), "max_mhz": fmax,
            "max_mv": int(mv), "gov_mode": "vf", "punti": len(pts)}


def apply_gpu_compat(minmhz, minmv, maxmhz, maxmv):
    """Old two-point call: move the ceiling and make sure the curve has that knot."""
    c = gov_leggi()
    pts = [list(p) for p in c.get("curva") or []]
    maxmhz, maxmv = int(maxmhz), int(maxmv)
    pts = [p for p in pts if p[0] != maxmhz]
    pts.append([maxmhz, maxmv])
    pts.sort()
    for i in range(1, len(pts)):
        if pts[i][1] < pts[i - 1][1]:
            pts[i][1] = pts[i - 1][1]
    c["curva"] = pts
    c["freq_max"] = maxmhz
    return gov_set(c).get("ok", False)


# ---- CPU --------------------------------------------------------------------
def _cpu_conf():
    cpu = {"frequency": 3500, "scale": 0, "max_temperature": 85}
    try:
        for line in _rd(OC_CONF).splitlines():
            m = re.match(r'(\w+)\s*=\s*(-?\d+)', line.strip())
            if m:
                cpu[m.group(1)] = int(m.group(2))
    except (OSError, ValueError):
        # no config yet, or a malformed line: keep the defaults set above
        pass
    return cpu


def _write_cpu_conf(mhz, scale, tmp):
    _wr(OC_CONF, "[overclock]\nfrequency = %d\nscale = %d\nmax_temperature = %d\n" % (mhz, scale, tmp))


# The backend is the only authority on what can be applied. bc250_limits.py is
# a plain list of integers next to bc250_apply.py, so it gets read instead of
# guessed: we used to accept 2000-4500 MHz against a floor of 3500, and every
# value below it was offered to the user and then refused without a word.
CPU_LIMITI_RIPIEGO = {"freq_min": 3500, "freq_max": 4500,
                      "scale_min": -50, "scale_max": 0,
                      "temp_min": 0, "temp_max": 100}

# The deepest undervolt we let anything reach. The backend allows -50, but -50
# at the stock clock froze the dev board outright and the suggestion sweep
# already stops at -40. Those ten steps buy nothing and cost a power cycle.
CPU_UV_SICURO = -40

# ⚠️ A dict, not a global that gets rebound. The cache used to be None and
# reassigned inside the function, and a scanner reads that as a global whose
# value nobody uses: it cannot follow a value that is only read on a later
# call. Filling a dict in place is the same cache and says so in the code.
_CPU_LIMITI = {}


def cpu_limiti():
    """What bc250_apply.py will actually accept, read once and kept."""
    if _CPU_LIMITI:
        return _CPU_LIMITI
    lim = dict(CPU_LIMITI_RIPIEGO)
    try:
        for nome, val in re.findall(r"(\w+)\s*=\s*(-?\d+)",
                                    _rd("%s/bc250_limits.py" % OC_DIR)):
            if nome in lim:
                lim[nome] = int(val)
    except OSError:
        # no backend installed: the fallback above is what it shipped with
        pass
    lim["scale_min"] = max(lim["scale_min"], CPU_UV_SICURO)
    _CPU_LIMITI.update(lim)
    return _CPU_LIMITI


def _cpu_fuori(mhz, scale, tmp):
    """Which knob sits outside the limits, as a code the panel can translate.
    None when everything fits."""
    lim = cpu_limiti()
    for campo, val, lo, hi in (
            ("frequency", mhz, lim["freq_min"], lim["freq_max"]),
            ("scale", scale, lim["scale_min"], lim["scale_max"]),
            ("max_temperature", tmp, lim["temp_min"], lim["temp_max"])):
        if not lo <= val <= hi:
            return {"campo": campo, "min": lo, "max": hi}
    return None


def _ultima_riga(testo):
    righe = [r.strip() for r in (testo or "").splitlines() if r.strip()]
    return righe[-1] if righe else ""


def apply_cpu_esito(mhz, scale, tmp):
    """Apply a CPU setting and say why if it did not go in.

    Returns (ok, reason). The reason is either a dict naming the knob that is
    out of range, which the panel turns into a sentence, or the backend's own
    last line. Plain False told the user nothing and read like a control that
    was not wired up.

    On refusal the previous configuration goes back on disk. This file is what
    bc250-smu-oc.service reads at boot, so a value the backend rejected would
    otherwise arm the next boot to fail as well, silently."""
    mhz, scale, tmp = int(mhz), int(scale), int(tmp)
    fuori = _cpu_fuori(mhz, scale, tmp)
    if fuori:
        return False, fuori
    if not os.path.exists("%s/bc250_apply.py" % OC_DIR):
        # issue #96: the backend was never installed on normal systems
        return False, "bc250_smu_oc is not installed: sudo apt install skillfish-smu-oc"
    _log("CPU: %d MHz, scale %d, limite %d C" % (mhz, scale, tmp))
    prima = _cpu_conf()
    _write_cpu_conf(mhz, scale, tmp)
    r = sh("python3 %s/bc250_apply.py --apply %s" % (OC_DIR, OC_CONF))
    if r.returncode == 0:
        return True, ""
    _write_cpu_conf(prima["frequency"], prima["scale"], prima["max_temperature"])
    return False, (_ultima_riga(r.stderr) or _ultima_riga(r.stdout)
                   or "bc250_apply.py failed")


def apply_cpu(mhz, scale, tmp):
    """Bool-only wrapper. Four callers branch on `if not apply_cpu(...)` and a
    tuple is always truthy, so the reason never travels through this name."""
    return apply_cpu_esito(mhz, scale, tmp)[0]


def _esito_cpu(ok, perche):
    """One shape for every CPU reply, so the panel always has something to say."""
    if ok:
        return {"ok": True}
    if isinstance(perche, dict):
        return {"ok": False, "fuori": perche}
    return {"ok": False, "err": perche}


def persist_cpu():
    # The unit comes with skillfish-smu-oc, guarded by skillfish-is-bc250.
    # `bc250_apply.py --install` would overwrite it with upstream's, which has
    # no guard: only the conf (written by apply_cpu_esito) and the enable here.
    sh("systemctl daemon-reload; systemctl enable bc250-smu-oc.service")


# An offline CPU has no topology/ in sysfs, so its core can't be read back.
# Every core_id seen while a CPU is online is kept here, and an offline CPU is
# placed with what was recorded. Without it the guess was n // 2: right on the
# BC-250, which pairs threads as (2k, 2k+1), wrong on a PC numbered otherwise,
# where "switch core 1 back on" then found no core 1 to switch on.
CPU_TOPO = "/var/lib/skillfish/cpu-topology.json"


def _topo_load():
    try:
        with open(CPU_TOPO) as f:
            return {int(k): int(v) for k, v in json.load(f).items()}
    except (OSError, ValueError, AttributeError):
        # first run, or a file we can't read: start from what is online now
        return {}


def _topo_save(topo):
    try:
        os.makedirs(os.path.dirname(CPU_TOPO), exist_ok=True)
        _wr(CPU_TOPO, json.dumps({str(k): v for k, v in sorted(topo.items())}) + NL)
    except OSError:
        # read-only or full disk: the guess below still works on the BC-250
        pass


def cpu_cores_get():
    cores = {}
    topo = _topo_load()
    seen = dict(topo)
    for p in sorted(glob.glob("/sys/devices/system/cpu/cpu[0-9]*")):
        n = os.path.basename(p)[3:]
        if not n.isdigit():
            continue
        n = int(n)
        try:
            core = int(_rd("%s/topology/core_id" % p).strip())
        except Exception:
            continue
        seen[n] = core
        online = True if not os.path.exists(p + "/online") else _rd(p + "/online").strip() == "1"
        cores.setdefault(core, {"core": core, "cpus": [], "online": False, "removable": True})
        cores[core]["cpus"].append({"cpu": n, "online": online, "removable": os.path.exists(p + "/online")})
        if online:
            cores[core]["online"] = True
        if not os.path.exists(p + "/online"):
            cores[core]["removable"] = False
    for p in sorted(glob.glob("/sys/devices/system/cpu/cpu[0-9]*")):
        n = os.path.basename(p)[3:]
        if not n.isdigit():
            continue
        n = int(n)
        if any(c["cpu"] == n for e in cores.values() for c in e["cpus"]):
            continue
        core = seen.get(n, n // 2)
        cores.setdefault(core, {"core": core, "cpus": [], "online": False, "removable": True})
        cores[core]["cpus"].append({"cpu": n, "online": False, "removable": True})
    if seen != topo:
        _topo_save(seen)
    smt = None
    try:
        smt = _rd("/sys/devices/system/cpu/smt/control").strip()
    except OSError:
        # no SMT control on this CPU: report unknown
        smt = None
    out = [cores[k] for k in sorted(cores)]
    for e in out:
        e["cpus"].sort(key=lambda c: c["cpu"])
    return {"ok": True, "cores": out, "nproc": os.cpu_count(), "smt": smt}


def cpu_cores_set(states):
    want = {}
    for s in states or []:
        try:
            want[int(s["core"])] = bool(s["online"])
        except (KeyError, TypeError, ValueError):
            # malformed entry: skip it, the rest of the request still applies
            pass
    if not want:
        return {"ok": False, "err": "nessun core specificato"}
    cur = cpu_cores_get()["cores"]
    if not any(want.get(e["core"], e["online"]) for e in cur):
        return {"ok": False, "err": "almeno un core deve restare acceso"}
    for e in cur:
        tgt = want.get(e["core"])
        if tgt is None:
            continue
        for c in e["cpus"]:
            if not c["removable"]:
                continue
            _wr("/sys/devices/system/cpu/cpu%d/online" % c["cpu"], "1" if tgt else "0")
    time.sleep(1)
    return cpu_cores_get()


def cpu_smt_set(on):
    p = "/sys/devices/system/cpu/smt/control"
    if not os.path.exists(p):
        return {"ok": False, "err": "SMT non controllabile su questo kernel"}
    _wr(p, "on" if on else "off")
    time.sleep(1)
    return cpu_cores_get()


# ---- CPU frequency governor (issue #95) -------------------------------------------
# The work is in skillfish-cpu-governor (skillfish-base), which the boot service
# and the udev rule call too: one implementation, three callers.
CPU_GOV = "/usr/local/bin/skillfish-cpu-governor"
ACPI_PSTATES = "/usr/local/bin/skillfish-acpi-pstates"


def _json_da(cmd, t=60):
    r = sh(cmd, t)
    try:
        return json.loads((r.stdout or "").strip().splitlines()[-1])
    except (ValueError, IndexError):
        return {"ok": False, "err": (r.stderr or r.stdout or "no answer").strip()[-300:]}


def cpu_gov_get():
    if not os.path.exists(CPU_GOV):
        return {"ok": False, "err": "manca skillfish-cpu-governor: aggiorna skillfish-base"}
    return _json_da(CPU_GOV + " status", 30)


def cpu_gov_set(gov):
    if not isinstance(gov, str) or not re.fullmatch(r"[a-z_]{2,20}", gov):
        return {"ok": False, "err": "governor non valido"}
    if not os.path.exists(CPU_GOV):
        return {"ok": False, "err": "manca skillfish-cpu-governor: aggiorna skillfish-base"}
    _log("governor CPU: %s" % gov)
    return _json_da("%s set %s" % (CPU_GOV, gov), 60)


def cpu_pstates_set(on):
    """Install or remove the ACPI P-state tables of a BC-250. Takes effect at
    the next boot: the tables are loaded by GRUB with the kernel."""
    if not os.path.exists(ACPI_PSTATES):
        return {"ok": False, "err": "manca skillfish-acpi-pstates"}
    r = sh("%s %s" % (ACPI_PSTATES, "enable" if on else "disable"), 180)
    _log("P-state ACPI: %s -> %s" % ("enable" if on else "disable", (r.stdout or "").strip()[-120:]))
    return {"ok": r.returncode == 0, "reboot": True, "out": (r.stdout or r.stderr or "").strip()[-400:]}


def core_unlock_get():
    try:
        n = len(os.sched_getaffinity(0))
    except (AttributeError, OSError):
        n = os.cpu_count() or 0
    return {"ok": True, "abilitato": os.path.exists(SEGNAPOSTO_CORE), "thread": n,
            "supportato": os.path.exists("/sys/bus/pci/devices/0000:00:00.0/config")}


def core_unlock_set(on):
    try:
        if on:
            os.makedirs("/etc/skillfish", exist_ok=True)
            _wr(SEGNAPOSTO_CORE, "# The presence of this file turns the 8-core unlock on.\n"
                "# Managed by SkillFishOS Control Center.\n")
        elif os.path.exists(SEGNAPOSTO_CORE):
            os.remove(SEGNAPOSTO_CORE)
    except OSError as e:
        return {"ok": False, "err": str(e)}
    d = core_unlock_get()
    d["riavvio"] = True
    return d


def thermal_guard(limit):
    limit = int(limit)
    if not (60 <= limit <= 100):
        return False
    os.makedirs("/etc/skillfish", exist_ok=True)
    _wr("/etc/skillfish/thermal-guard.conf",
        "# Written by SkillFishOS Control Center. Above this temperature the CPU\n"
        "# is slowed by 100 MHz at a time.\nLIMITE=%d\n" % limit)
    sh("systemctl enable --now skillfish-thermal-guard.service; systemctl restart skillfish-thermal-guard.service")
    return True


def thermal_guard_get():
    m = re.search(r"LIMITE=(\d+)", _rd("/etc/skillfish/thermal-guard.conf") if os.path.exists(
        "/etc/skillfish/thermal-guard.conf") else "")
    return {"ok": True, "limite": int(m.group(1)) if m else None,
            "attiva": _unit("is-active", "skillfish-thermal-guard.service")}


# ---- compute units ----------------------------------------------------------
def cu_get():
    try:
        j = json.loads(sh("/usr/local/bin/skillfish-cu get", 30).stdout)
    except Exception:
        return {}
    # Whether a mapping is kept for the next boot is asked of the tool, which
    # owns the file, rather than by looking for the file from here.
    try:
        b = json.loads(sh("/usr/local/bin/skillfish-cu boot-get", 30).stdout)
        j["keep_boot"] = bool(b.get("keep"))
        j["boot_rows"] = b.get("rows") or []
    except Exception:
        j["keep_boot"] = False
        j["boot_rows"] = []
    return j


ORDINE_CU = ("0.0", "0.1", "1.0", "1.1")


def _cu_righe_valide(rows):
    """Four masks, each one a real one and none of them empty.

    A row with no WGP has nothing to run on, and the tool would quietly raise
    it to one pair. Saying no here means the window can explain it instead of
    showing a number the user did not choose."""
    if not isinstance(rows, list) or len(rows) != 4:
        return "maschere-storte"
    for x in rows:
        if not isinstance(x, int) or isinstance(x, bool) or not 0 <= x <= 31:
            return "maschere-storte"
    if any(x == 0 for x in rows):
        return "riga-vuota"
    return None


def cu_apply(rows):
    try:
        if not os.path.exists("/usr/local/bin/umr"):
            return {"ok": False, "err": "manca /usr/local/bin/umr, senza il quale le CU non si possono instradare"}
        if not os.path.exists("/usr/local/share/umr/database/pci.did"):
            return {"ok": False, "err": "manca il database dei registri di umr: reinstalla skillfish-tuner"}
        storto = _cu_righe_valide(rows)
        if storto:
            # a missing fourth row used to be filled in with 0x07, which is the
            # floor that turned out not to exist: better to refuse than to
            # invent a mapping nobody asked for
            return {"ok": False, "motivo": storto}
        r = list(rows)
        _log("CU: maschere %s" % r)
        p = sh("/usr/local/bin/skillfish-cu set-rows %s" % (" ".join(str(x) for x in r)))
        j = cu_get()
        atteso = dict(zip(ORDINE_CU, r))
        if p.returncode != 0 or j.get("rows") != atteso:
            # the card is the authority, not the exit code: a write that went
            # nowhere used to come back as success. From PR #80.
            motivo = (getattr(p, "stderr", "") or "").strip() or (p.stdout or "").strip()
            return {"ok": False, "err": motivo or "skillfish-cu e' uscito con %d" % p.returncode,
                    "active": j.get("active_cu"), "rows": j.get("rows")}
        return {"ok": True, "active": j.get("active_cu"), "rows": j.get("rows")}
    except Exception as e:
        return {"ok": False, "err": str(e)}


def cu_keep_boot(on, rows):
    """Keep this mapping at the next boot, or stop keeping one.

    Only a mapping that is actually on the card can be kept: saving something
    that was never applied would arm the next boot with a routing nobody has
    seen work. The idea, and the check, are from tom lima's PR #80."""
    # The unit is what reads the saved mapping at boot. Issue #98: on systems
    # installed from an ISO it was never enabled, so a saved mapping, and the
    # plain 40 CU, were never applied. Enabled here whatever the choice: with
    # nothing saved it switches all 40 on.
    sh("systemctl enable skillfish-cu.service", 30)
    if not on:
        p = sh("/usr/local/bin/skillfish-cu boot-clear", 30)
        return {"ok": p.returncode == 0, **{k: v for k, v in cu_get().items() if k != "ok"}}
    storto = _cu_righe_valide(rows)
    if storto:
        return {"ok": False, "motivo": storto}
    viva = cu_get().get("rows")
    if viva != dict(zip(ORDINE_CU, rows)):
        return {"ok": False, "motivo": "non-applicata"}
    p = sh("/usr/local/bin/skillfish-cu boot-save %s" % " ".join(str(x) for x in rows), 30)
    _log("CU: tenute al boot %s" % rows)
    return {"ok": p.returncode == 0, **{k: v for k, v in cu_get().items() if k != "ok"}}


def cu_test():
    VKPEAK = find_vkpeak()
    if not VKPEAK:
        return {"ok": False, "err": "vkpeak is not installed: sudo apt install skillfish-vkpeak"}
    vdir = os.path.dirname(VKPEAK)

    def vk():
        # 12 and not 22: fp32-scalar is on screen within 8 seconds, measured on
        # the dev board, and the sweep now has twenty steps to get through.
        r = sh("cd %s && stdbuf -oL -eL timeout 12 ./vkpeak" % vdir, t=25)
        m = re.search(r'fp32-scalar\s*=\s*([\d.]+)', r.stdout)
        return (float(m.group(1)) if m else 0.0), r.returncode

    def alive():
        return "BC-250" in sh("timeout 12 vulkaninfo --summary 2>/dev/null | grep -m1 deviceName", t=20).stdout

    def errs():
        s = sh("dmesg --since '16 seconds ago' 2>/dev/null | grep -ciE 'amdgpu.*(fault|timeout|reset|hang|recover|failed)'").stdout.strip()
        return int(s) if s.isdigit() else 0

    stato = cu_get()
    cur = stato.get("rows", {})
    order = list(ORDINE_CU)
    # The lowest pair is what the other rows run on while one row is under
    # test. It used to be 0x07, three pairs, which is exactly the range the
    # test could never examine.
    fondo = 1 << 0
    res = []

    def rimetti():
        """The mapping the user had before any of this started.

        ⚠️ In a finally, because it used to be the last statement of the
        function: anything that raised part way through twenty steps left the
        board on one pair per row, eight CU out of forty, and said nothing."""
        g0 = int  # keeps the lambda-free shape readable
        sh("/usr/local/bin/skillfish-cu set-rows %d %d %d %d"
           % tuple(g0(cur.get(k, 31)) for k in order))

    try:
        sh("/usr/local/bin/skillfish-cu set-rows %d %d %d %d" % (fondo, fondo, fondo, fondo))
        time.sleep(0.5)
        base, _ = vk()
        for rk in order:
            for wgp in range(5):
                rows = {k: fondo for k in order}
                rows[rk] = (1 << wgp)
                sh("/usr/local/bin/skillfish-cu set-rows %d %d %d %d" % (rows["0.0"], rows["0.1"], rows["1.0"], rows["1.1"]))
                time.sleep(0.5)
                g, rc = vk()
                e = errs()
                al = alive()
                verdict = "FAIL" if (not al or e > 0) else ("N/A" if g <= 0 else "OK")
                res.append({"row": rk, "wgp": wgp, "cu": "%d-%d" % (wgp * 2, wgp * 2 + 1),
                            "gflops": round(g), "errors": e, "verdict": verdict})
                time.sleep(1)
        sh("/usr/local/bin/skillfish-cu max")
        time.sleep(0.5)
        # ⚠️ timeout, like every other bench here. Without one, a subprocess
        # deadline kills the shell and leaves ./vkpeak orphaned on the GPU.
        rf = sh("cd %s && stdbuf -oL -eL timeout 20 ./vkpeak" % vdir, t=35)
        mf = re.search(r'fp32-scalar\s*=\s*([\d.]+)', rf.stdout)
        full = round(float(mf.group(1)) if mf else 0.0)
        full_err = errs()
    finally:
        rimetti()
    return {"ok": True, "baseline": round(base), "results": res,
            "bad": sum(1 for x in res if x["verdict"] == "FAIL"),
            "na": sum(1 for x in res if x["verdict"] == "N/A"), "full40": full, "full40_err": full_err}


# ---- VRAM -----------------------------------------------------------------------
def vram_get():
    if os.path.exists(MEMCFG):
        m = re.search(r'UMA_SIZE=(\d+)', sh(MEMCFG + ' get').stdout)
        if m:
            return int(m.group(1))
    return 0


def set_vram(mb):
    if not os.path.exists(MEMCFG):
        return False
    mb = max(256, int(mb))
    nostro = MEMCFG.endswith("skillfish-memcfg")
    cmd = "%s set %d" % (MEMCFG, mb) if nostro else "%s UMA_SIZE %d" % (MEMCFG, mb)
    _log("VRAM: %d MB" % mb)
    if sh(cmd).returncode != 0:
        return False
    letto = sh("%s %s" % (MEMCFG, "get" if nostro else "")).stdout or ""
    m = re.search(r"UMA_SIZE\s*[=:]?\s*(\d+)", letto)
    return bool(m) and int(m.group(1)) == mb


# ---- scheduler ----------------------------------------------------------------
def _testo(p):
    try:
        return _rd(p).strip()
    except OSError:
        return ""


def scx_get():
    supportato = os.path.isdir("/sys/kernel/sched_ext") and os.path.exists(SCX_BIN)
    conta = _testo(SCX_CONTA)
    return {"ok": True, "supportato": supportato, "installato": os.path.exists(SCX_BIN),
            "kernel": os.path.isdir("/sys/kernel/sched_ext"),
            "abilitato": _unit("is-enabled", SCX_PATH_UNIT),
            "attivo": _unit("is-active", SCX_SERVICE),
            "caricato": _testo(SCX_STATO) == "enabled", "nome": _testo(SCX_OPS),
            "espulsioni": int(conta) if conta.isdigit() else 0,
            "in_gioco": os.path.exists("/run/skillfish/gamemode/attivo")}


def scx_set(on):
    if not os.path.isdir("/sys/kernel/sched_ext"):
        return {"ok": False, "err": "questo kernel non ha sched_ext"}
    if not os.path.exists(SCX_BIN):
        return {"ok": False, "err": "manca skillfish-scx"}
    if on:
        r = sh("systemctl enable --now " + SCX_PATH_UNIT, 30)
    else:
        r = sh("systemctl disable --now %s; systemctl stop %s" % (SCX_PATH_UNIT, SCX_SERVICE), 30)
    if r.returncode != 0:
        return {"ok": False, "err": "systemctl ha rifiutato (%d)" % r.returncode}
    return scx_get()


def scx_azzera():
    _wr(SCX_CONTA, "0\n")
    return scx_get()


# ---- our Mesa as the system driver -------------------------------------------
# ⚠️ THE WORK IS NOT DONE HERE ANY MORE. Up to 26.09.3 this pair of functions
# diverted one file, /usr/lib/x86_64-linux-gnu/libvulkan_radeon.so, which covered
# Vulkan and left the desktop on Debian's OpenGL. From 26.09.4 the package ships
# the whole driver and skillfish-mesa owns the switching, host and launchers
# together, so there is one implementation instead of two that can disagree.
# These two stay because the window and the web page call them by name.
def mesa_sistema_get():
    r = sh("%s json" % MESA_CLI, 30)
    if r.returncode == 0 and r.stdout.strip():
        try:
            return json.loads(r.stdout)
        except ValueError:
            pass
    # the switch is missing or answered nonsense: say what can still be known
    return {"ok": True, "attivo": False, "installata": os.path.exists(MESA_NOSTRA),
            "kernel_nostro": "skillfishos" in os.uname().release,
            "versione": sh("dpkg-query -W -f='${Version}' %s 2>/dev/null" % MESA_PKG, 10).stdout.strip(),
            "motivo": "manca %s" % MESA_CLI}


def mesa_sistema_set(on):
    """Hand it to the switch: the host half and the two launchers move together."""
    if not os.path.exists(MESA_CLI):
        return {"ok": False, "err": "la nostra Mesa non e' installata (%s)" % MESA_PKG}
    _log("Mesa di sistema: %s" % ("nostra" if on else "di serie"))
    r = sh("%s %s" % (MESA_CLI, "on" if on else "off"), 120)
    st = mesa_sistema_get()
    if r.returncode != 0:
        st["ok"] = False
        st["err"] = (r.stdout or r.stderr).strip() or st.get("motivo") or "non riuscito"
    return st


# ---- other helpers, reached through this one -----------------------------------
def kernel(azione, kv):
    if azione not in ("default", "once", "uninstall") or not re.match(r"^[A-Za-z0-9._+-]+$", kv or ""):
        return {"ok": False, "err": "richiesta non valida"}
    r = sh("%s %s %s" % (KERNEL_HELPER, azione, kv), 900)
    return {"ok": r.returncode == 0, "out": (r.stdout or "").strip(), "err": (r.stderr or "").strip()}


def snapshot(argomenti, attesa=600):
    if not isinstance(argomenti, list) or not all(isinstance(a, str) for a in argomenti):
        return {"ok": False, "err": "richiesta non valida"}
    if not os.path.exists(SNAP_HELPER):
        return {"ok": False, "err": "manca skillfish-snapshots"}
    try:
        p = subprocess.run([SNAP_HELPER] + argomenti, capture_output=True, text=True, timeout=attesa)
        return {"ok": p.returncode == 0, "out": (p.stdout or "").strip(), "err": (p.stderr or "").strip(),
                "rc": p.returncode}
    except Exception as e:
        return {"ok": False, "err": str(e)}


def manutenzione(argomenti):
    if not isinstance(argomenti, list) or not all(isinstance(a, str) for a in argomenti):
        return {"ok": False, "err": "richiesta non valida"}
    if not os.path.exists(SNAP_MANUT):
        return {"ok": False, "err": "manca skillfish-btrfs-manutenzione"}
    try:
        p = subprocess.run([SNAP_MANUT] + argomenti, capture_output=True, text=True, timeout=600)
        return {"ok": p.returncode == 0, "out": (p.stdout or "").strip(), "err": (p.stderr or "").strip()}
    except Exception as e:
        return {"ok": False, "err": str(e)}


def gddr6(azione):
    """Start or stop the capped GDDR6 reading, through its own helper.

    ⚠️ THE CEILING AND THE COOLDOWN ARE NOT HERE. skillfish-gddr6-helper owns
    both, because polling those sensors wedges the SMU - the chip the V/F
    governor depends on - and a second place deciding how long a session may run
    is a second place to get it wrong. This only forwards the two verbs.

    `stato` is deliberately absent: it needs no privileges and the page reads it
    on its own, so opening the Monitor never asks for a password.
    """
    if azione not in ("avvia", "ferma"):
        return {"ok": False, "errore": "azione: avvia | ferma"}
    try:
        p = subprocess.run(["/usr/local/bin/skillfish-gddr6-helper", azione],
                           capture_output=True, text=True, timeout=45)
        return json.loads((p.stdout or "{}").strip().splitlines()[-1])
    except Exception as errore:
        return {"ok": False, "errore": str(errore)[:200]}


def ventola(azione, dati):
    """The fan helper rebuilds the configuration field by field: we pass it on."""
    if azione not in ("scrivi", "etichette", "prova"):
        return {"ok": False, "err": "azione non valida"}
    if not os.path.exists(FAN_HELPER):
        return {"ok": False, "err": "manca skillfish-fan"}
    try:
        p = subprocess.run([FAN_HELPER, azione], input=json.dumps(dati), capture_output=True,
                           text=True, timeout=200)
        out = {"ok": p.returncode == 0, "err": (p.stderr or p.stdout or "").strip()}
        try:
            out["dati"] = json.loads(p.stdout)
        except json.JSONDecodeError:
            # not every action prints JSON: out["ok"]/out["err"] already say enough
            pass
        return out
    except Exception as e:
        return {"ok": False, "err": str(e)}


def servizio(azione, unit):
    """enable/disable/start/stop/restart for a short allow-list of our units."""
    permessi = ("skillfish-unsloth.service", "skillfish-dashboard.service", "skillfish-fand.service",
                "skillfish-thermal-guard.service", "skillfish-vf-governor.service",
                "skillfish-scx.path", "skillfish-cu.service", "bc250-smu-oc.service",
                "skillfish-gaming-mode.service")
    if unit not in permessi or azione not in ("enable --now", "disable --now", "enable", "disable", "start", "stop", "restart"):
        return {"ok": False, "err": "non permesso"}
    r = sh("systemctl %s %s" % (azione, unit), 120)
    return {"ok": r.returncode == 0, "err": (r.stderr or "").strip(),
            "attivo": _unit("is-active", unit), "abilitato": _unit("is-enabled", unit)}


UNSLOTH_DEFAULT = "/etc/default/skillfish-unsloth"
UNSLOTH_KEY_DASH = "/etc/skillfish/unsloth.key"


def unsloth_conf():
    """Bind address and parallel slots of the engine, from its environment file."""
    out = {"ok": True, "bind": "127.0.0.1", "parallel": 4}
    try:
        for line in _rd(UNSLOTH_DEFAULT).splitlines():
            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["attivo"] = _unit("is-active", "skillfish-unsloth.service")
    out["abilitato"] = _unit("is-enabled", "skillfish-unsloth.service")
    return out


def unsloth_conf_set(lan, parallel):
    """Write the environment file the unit reads, restart the engine if it runs."""
    try:
        parallel = max(1, min(64, int(parallel)))
    except (TypeError, ValueError):
        parallel = 4
    bind = "0.0.0.0" if lan else "127.0.0.1"
    testo = ("# 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" % (bind, parallel))
    try:
        with open(UNSLOTH_DEFAULT, "w") as f:
            f.write(testo)
    except OSError as e:
        return {"ok": False, "err": str(e)}
    _log("unsloth: bind %s, parallel %d" % (bind, parallel))
    if _unit("is-active", "skillfish-unsloth.service") == "active":
        sh("systemctl restart skillfish-unsloth.service", 120)
    return unsloth_conf()


def cluster(azione, ip="", utente="", password=""):
    """Il cluster AI: elenco, telemetria, aggiunta, avvio dei nodi.

    ⚠️ Passa da qui e non dalla finestra perche' serve root: la chiave ssh del
    cluster sta in /etc/skillfish e l'elenco delle schede pure.
    """
    cmd = ["/usr/local/bin/skillfish-cluster"]
    if azione in ("stato", "elenco", "avvia", "ferma"):
        cmd.append(azione)
    elif azione == "aggiungi":
        if not ip or not utente:
            return {"ok": False, "errore": "indirizzo o utente mancante"}
        cmd += ["aggiungi", ip, utente, password]
    elif azione == "togli":
        cmd += ["togli", ip]
    else:
        return {"ok": False, "errore": "azione sconosciuta"}
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
    try:
        d = json.loads(r.stdout or "{}")
    except ValueError:
        return {"ok": False, "errore": (r.stderr or r.stdout or "").strip()[-300:]}
    d.setdefault("ok", r.returncode == 0)
    # ⚠️ La password non finisce nel registro: si scrive cosa e' stato fatto,
    # non con cosa.
    if azione in ("aggiungi", "togli", "avvia", "ferma"):
        _log("cluster: %s %s" % (azione, ip or ""))
    return d


def ai_livello(livello="", modello=""):
    """Scrive il livello scelto e il modello del motore nudo.

    ⚠️ Passa da root perche' il file sta in /etc/skillfish, ma NON accende
    niente: accendere e' un'altra azione, con la sua domanda. Cambiare la
    scelta mentre la modalita' e' gia' accesa non ha effetto fino al prossimo
    giro, ed e' giusto cosi': spegnere il motore sotto i piedi di chi sta
    chiedendo qualcosa al modello sarebbe peggio.
    """
    # ⚠️ Stesso controllo del Remote Manager: l'elenco chiuso vale anche qui,
    # anche se qui dall'altra parte c'e' la nostra finestra. Un helper che gira
    # da root non si fida di chi lo chiama.
    scelto = ""
    for noto in ("studio", "motore", "motore-api"):
        if noto == livello:
            scelto = noto
            break
    if livello and not scelto:
        return {"ok": False, "errore": "livello sconosciuto"}
    if modello and not os.path.isfile(modello):
        return {"ok": False, "errore": "modello inesistente"}
    cmd = ["/usr/local/bin/skillfish-llama", "imposta"]
    if livello:
        cmd.append("livello=" + scelto)
    if modello:
        cmd.append("modello=%s" % modello)
    if len(cmd) == 2:
        return {"ok": False, "errore": "niente da cambiare"}
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    try:
        d = json.loads(r.stdout or "{}")
    except ValueError:
        return {"ok": False, "errore": (r.stderr or r.stdout or "").strip()[-200:]}
    _log("ai: livello %s" % (livello or "(invariato)"))
    return d


def ai_mode(azione):
    """Accende o spegne la modalita' AI: il desktop si ferma, il modello resta.

    ⚠️ Accendendola si chiude quello che l'utente ha aperto, senza salvare. La
    domanda la fa la finestra; qui si esegue e basta.
    """
    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 unsloth_password_reset(scelta=""):
    """Ruota la password di Studio e restituisce quella nuova.

    ⚠️ Il binario sta nella HOME dell'utente che ha installato Studio, non in
    /usr: va eseguito COME QUELL'UTENTE, altrimenti scrive un auth.db di root
    accanto a quello vero e Studio continua a rifiutare la password.

    ⚠️ La password nuova la stampa il comando e non la salviamo da nessuna
    parte: chi la vuole se la scrive. Un file con dentro la password di Studio
    sarebbe una password in chiaro sul disco, che e' esattamente quello che
    stiamo cercando di non fare.
    """
    utente = _unsloth_utente()
    if not utente:
        return {"ok": False, "errore": "no Unsloth installation found"}
    casa = pwd.getpwnam(utente).pw_dir
    binario = os.path.join(casa, ".unsloth", "studio", "unsloth_studio", "bin", "unsloth")
    if not os.access(binario, os.X_OK):
        return {"ok": False, "errore": "%s non e' eseguibile" % binario}
    r = subprocess.run(["runuser", "-u", utente, "--", binario, "studio", "reset-password"],
                       capture_output=True, text=True, timeout=120)
    testo = (r.stdout or "") + (r.stderr or "")
    if r.returncode != 0:
        return {"ok": False, "errore": testo.strip()[-400:] or "reset-password non riuscito"}
    # ⚠️ La password sta DOPO i due punti, su una riga che di spazi ne ha:
    #     New password for 'unsloth': PulpBalconyThrongDiabetic
    # La prima stesura cercava la riga piu' lunga senza spazi e non trovava
    # niente, quindi la password scelta dall'utente non veniva mai impostata.
    nuova = ""
    for riga in testo.splitlines():
        riga = riga.strip()
        if "password" in riga.lower() and ":" in riga:
            cand = riga.split(":", 1)[1].strip().strip("'\"`")
            if cand and " " not in cand and len(cand) >= 8:
                nuova = cand
                break
    if not scelta:
        # ⚠️ Anche qui la chiave va rifatta: il reset l'ha revocata comunque, e
        # senza questa riga il Control Center resta con una chiave morta in
        # mano senza accorgersene.
        chiave, _ = _studio_nuova_chiave(nuova)
        if chiave:
            unsloth_key_set(chiave)
        _log("unsloth: password di Studio ruotata (casuale)")
        return {"ok": True, "password": nuova, "scelta": False,
                "chiave": chiave, "chiave_rifatta": bool(chiave),
                "uscita": testo.strip()[-400:]}

    # ⚠️ Il secondo passo: da quella casuale a quella che l'utente ha scelto.
    # `reset-password` non prende argomenti (verificato col suo --help), quindi
    # e' l'unica strada. Se qui qualcosa va storto NON si fallisce: la casuale
    # e' valida, e la finestra la mostra con il pulsante per copiarla. Meglio
    # una password brutta che si puo' copiare che una scheda in cui non si
    # entra piu'.
    if not nuova:
        return {"ok": True, "password": "", "scelta": False,
                "errore": "password nuova non leggibile dall'uscita del comando",
                "uscita": testo.strip()[-400:]}
    err = _studio_cambia_password(nuova, scelta)
    if err:
        # La password buona adesso e' quella casuale: la chiave si rifa' con
        # quella, non con la scelta che Studio ha rifiutato.
        chiave, _ = _studio_nuova_chiave(nuova)
        if chiave:
            unsloth_key_set(chiave)
        _log("unsloth: password ruotata, ma il cambio a quella scelta non e' riuscito")
        return {"ok": True, "password": nuova, "scelta": False, "errore": err,
                "chiave": chiave}
    # ⚠️ IL RESET HA REVOCATO ANCHE LE CHIAVI API: lo dice l'uscita del comando.
    # Senza rifarla, dopo questo smettono di funzionare insieme il Control
    # Center, il Remote Manager e qualunque editor collegato - e nessuno
    # collegherebbe la cosa al cambio di password.
    chiave, err_k = _studio_nuova_chiave(scelta)
    if chiave:
        unsloth_key_set(chiave)
    _log("unsloth: password impostata da chi usa la macchina, chiave API rifatta"
         if chiave else "unsloth: password impostata, chiave API DA RIFARE: %s" % err_k)
    # ⚠️ La chiave torna alla finestra, che la salva in casa dell'utente.
    # unsloth_key_set() scrive SOLO la copia che legge il Remote Manager: se ci
    # si ferma li', il Control Center continua a usare quella vecchia e revocata
    # e mostra schede vuote senza dire perche'.
    return {"ok": True, "password": "", "scelta": True,
            "chiave": chiave,
            "chiave_rifatta": bool(chiave),
            "errore_chiave": "" if chiave else err_k}


def _studio_nuova_chiave(password):
    """Una chiave API nuova dopo il reset. Restituisce (chiave, motivo)."""
    import json as _json
    import urllib.request
    try:
        def chiama(rotta, dati, token=""):
            req = urllib.request.Request(
                "http://127.0.0.1:8888" + rotta,
                data=_json.dumps(dati).encode("utf-8"),
                headers={"Content-Type": "application/json",
                         **({"Authorization": "Bearer " + token} if token else {})})
            with urllib.request.urlopen(req, timeout=20) as r:
                return _json.loads(r.read().decode("utf-8"))

        # ⚠️ Si riprova: dopo un reset il rate limit di Studio puo' restare
        # chiuso fino a un minuto anche con la password giusta, e un solo
        # tentativo fa credere che la password non vada.
        import time
        tok = ""
        for _ in range(12):
            try:
                r = chiama("/api/auth/login",
                           {"username": "unsloth", "password": password})
                tok = r.get("access_token") or ""
                if tok:
                    break
            except Exception:
                # ⚠️ Qui si ingoia di proposito: e' il giro dei tentativi, e
                # Studio appena riavviato risponde in tutti i modi sbagliati
                # prima di rispondere bene. Quello che conta e' l'ultimo giro,
                # e se falliscono tutti lo dice `ultimo` piu' sotto.
                pass
            time.sleep(5)
        if not tok:
            return "", "non entro in Studio con la password nuova"
        r2 = chiama("/api/auth/api-keys", {"name": "SkillFishOS"}, tok)
        k = r2.get("key") or r2.get("api_key") or ""
        return (k, "") if k else ("", "Studio non ha restituito una chiave")
    except Exception as e:
        return "", str(e)


def _studio_cambia_password(vecchia, nuova):
    """Da quella casuale a quella scelta, passando dall'API di Studio.

    Restituisce "" se e' andata, altrimenti il motivo. ⚠️ Studio ascolta su
    127.0.0.1 salvo che l'utente non l'abbia aperto: si passa sempre da li',
    anche quando e' aperto in rete, perche' e' l'unico indirizzo che c'e'
    sempre.
    """
    import json as _json
    import urllib.error
    import urllib.request

    def chiama(rotta, dati, token=""):
        req = urllib.request.Request(
            "http://127.0.0.1:8888" + rotta,
            data=_json.dumps(dati).encode("utf-8"),
            headers={"Content-Type": "application/json",
                     **({"Authorization": "Bearer " + token} if token else {})})
        with urllib.request.urlopen(req, timeout=20) as r:
            return _json.loads(r.read().decode("utf-8"))

    # ⚠️ Subito dopo un reset Studio puo' rifiutare anche la password giusta:
    # lo dice il comando stesso, "repeated failed logins can hold the rate
    # limit shut for up to a minute". Si riprova per un minuto invece di
    # dichiarare un guasto che non c'e'.
    import time
    tok = ""
    ultimo = ""
    for _ in range(12):
        try:
            r = chiama("/api/auth/login", {"username": "unsloth", "password": vecchia})
            tok = r.get("access_token") or ""
            if tok:
                break
            ultimo = "la password appena generata non e' stata accettata"
        except Exception as e:
            ultimo = "Studio non risponde: %s" % e
        time.sleep(5)
    if not tok:
        return ultimo or "non sono riuscito a entrare in Studio"
    try:
        chiama("/api/auth/change-password",
               {"current_password": vecchia, "new_password": nuova}, tok)
    except urllib.error.HTTPError as e:
        corpo = ""
        try:
            corpo = e.read().decode("utf-8", "replace")[:200]
        except Exception:
            # il corpo della risposta e' un di piu': se non si legge resta il
            # codice HTTP, che e' gia' la meta' del messaggio d'errore.
            pass
        return "Studio ha rifiutato la password scelta: %s %s" % (e.code, corpo)
    except Exception as e:
        return str(e)
    return ""


def _unsloth_utente():
    """Chi ha Studio installato in casa. Stesso criterio di skillfish-unsloth."""
    for u in pwd.getpwall():
        if 1000 <= u.pw_uid < 65534 and os.path.isdir(os.path.join(u.pw_dir, ".unsloth")):
            return u.pw_name
    return ""


def unsloth_key_set(key):
    """The copy of the API key the Remote Manager reads (root, 0600)."""
    key = (key or "").strip()
    try:
        if not key:
            if os.path.exists(UNSLOTH_KEY_DASH):
                os.remove(UNSLOTH_KEY_DASH)
            return {"ok": True}
        if not re.match(r"^sk-[A-Za-z0-9._-]{8,200}$", key):
            return {"ok": False, "err": "chiave non valida"}
        os.makedirs(os.path.dirname(UNSLOTH_KEY_DASH), exist_ok=True)
        fd = os.open(UNSLOTH_KEY_DASH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(key + "\n")
        return {"ok": True}
    except OSError as e:
        return {"ok": False, "err": str(e)}


def gtt(mb):
    """The GTT limit of the AI panel goes through skillfish-gtt."""
    if not os.path.exists("/usr/local/bin/skillfish-gtt"):
        return {"ok": False, "err": "manca skillfish-gtt"}
    r = sh("/usr/local/bin/skillfish-gtt %d" % int(mb), 120)
    return {"ok": r.returncode == 0, "out": (r.stdout or r.stderr or "").strip()}


# ---- benchmarks (kept from the Tuner) --------------------------------------------
def _govs_get():
    out = {}
    for p in glob.glob("/sys/devices/system/cpu/cpu*/cpufreq/scaling_governor"):
        try:
            out[p] = _rd(p).strip()
        except OSError:
            # this CPU went offline between the glob and the read: skip it
            pass
    return out


def bench_cpu(secs=60):
    nth = os.cpu_count() or 6
    import threading
    prev_govs = _govs_get()
    for p in prev_govs:
        try:
            _wr(p, "performance")
        except OSError:
            # this CPU refused the governor change: benchmark it as-is
            pass
    samp = {"minf": 99999, "maxt": 0}
    stop = threading.Event()

    def mon():
        deadline = time.monotonic() + 8
        while not stop.is_set() and time.monotonic() < deadline:
            time.sleep(0.5)
        while not stop.is_set():
            f = cpu_min_freq()
            t = temp("k10temp")
            if 0 < f < samp["minf"]:
                samp["minf"] = f
            if t > samp["maxt"]:
                samp["maxt"] = t
            time.sleep(2)
    th = threading.Thread(target=mon, daemon=True)
    th.start()
    r = sh("sysbench cpu --threads=%d --time=%d --cpu-max-prime=20000 run" % (nth, secs), t=secs + 30)
    stop.set()
    th.join(timeout=3)
    for p, g in prev_govs.items():
        try:
            _wr(p, g)
        except OSError:
            # this CPU went offline during the run: nothing left to restore
            pass
    m = re.search(r'events per second:\s*([\d.]+)', r.stdout)
    eps = float(m.group(1)) if m else 0
    minf = samp["minf"] if samp["minf"] < 99999 else cpu_min_freq()
    return {"score": round(eps, 1), "unit": "ev/s", "min_mhz": minf,
            "temp": samp["maxt"] or temp("k10temp"), "ok": r.returncode == 0 and eps > 0}


def bench_gpu():
    VKPEAK = find_vkpeak()
    if not VKPEAK:
        return {"score": 0, "unit": "GFLOPS", "ok": False, "err": "vkpeak is not installed: sudo apt install skillfish-vkpeak"}
    r = sh("cd %s && ./vkpeak" % os.path.dirname(VKPEAK), t=150)
    m = re.search(r'fp32-scalar\s*=\s*([\d.]+)', r.stdout)
    g = float(m.group(1)) if m else 0
    return {"score": round(g, 0), "unit": "GFLOPS", "temp": temp("amdgpu"), "ok": r.returncode == 0 and g > 0}


def test_cpu(mhz, scale, tmp):
    prev = _cpu_conf()
    if not apply_cpu(mhz, scale, tmp):
        _write_cpu_conf(prev["frequency"], prev["scale"], prev["max_temperature"])
        return {"ok": False, "phase": "apply", "err": "applicazione fallita"}
    # while the candidate is benched the LAST GOOD values stay on disk: a hard
    # freeze mid-bench must not boot into the unstable candidate
    _write_cpu_conf(prev["frequency"], prev["scale"], prev["max_temperature"])
    b = bench_cpu()
    stable = b["ok"] and b["min_mhz"] >= (int(mhz) - 200)
    if not stable:
        apply_cpu(prev["frequency"], prev["scale"], prev["max_temperature"])
        return {"ok": False, "applied": False, "bench": b,
                "err": "Instabile/throttle: %d MHz sotto carico (target %d). Ripristinato." % (b["min_mhz"], int(mhz))}
    _write_cpu_conf(int(mhz), int(scale), int(tmp))
    return {"ok": True, "applied": True, "bench": b}


def cpu_volatile(mhz, scale, tmp):
    """Apply a CPU setting WITHOUT leaving it on disk: the Control Center runs
    its own load loop (with a countdown and a Stop button) and calls this per
    step; the last good values keep booting if the board dies mid-test."""
    prev = _cpu_conf()
    ok, perche = apply_cpu_esito(mhz, scale, tmp)
    _write_cpu_conf(prev["frequency"], prev["scale"], prev["max_temperature"])
    return _esito_cpu(ok, perche)


COREUNLOCK_EFI = "/usr/local/bin/skillfish-coreunlock-efi"


def coreunlock_efi_get():
    if not os.path.exists(COREUNLOCK_EFI):
        return {"ok": True, "supportato": False}
    r = sh("%s stato" % COREUNLOCK_EFI, 20)
    out = r.stdout or ""
    voce = re.search(r'voce di avvio: (Boot[0-9A-F]{4})', out)
    ordine = re.search(r'ordine: ([0-9A-F,]+)', out)
    primo = bool(voce and ordine and ordine.group(1).split(",")[0] == voce.group(1)[4:])
    return {"ok": True, "supportato": True, "installato": bool(voce), "primo": primo,
            "programma": "presente" in out, "secure_boot": os.path.exists("/sys/firmware/efi/efivars") and any(
                n.startswith("SecureBoot-") for n in os.listdir("/sys/firmware/efi/efivars"))}


def coreunlock_efi_set(on):
    if not os.path.exists(COREUNLOCK_EFI):
        return {"ok": False, "err": "manca skillfish-coreunlock-efi"}
    r = sh("%s %s" % (COREUNLOCK_EFI, "installa" if on else "togli"), 30)
    _log("sblocco EFI: %s -> %s" % ("installa" if on else "togli", (r.stdout or "").strip()[:120]))
    return {"ok": r.returncode == 0, "out": (r.stdout or "").strip(), **coreunlock_efi_get()}


def suggest_uv(mhz):
    prev = _cpu_conf()
    best, s = 0, 0
    while s > -20:
        if not apply_cpu(mhz, s, prev["max_temperature"]):
            break
        _write_cpu_conf(prev["frequency"], prev["scale"], prev["max_temperature"])
        b = bench_cpu(12)
        if b["ok"] and b["min_mhz"] >= (int(mhz) - 200):
            best = s
            s -= 2
        else:
            break
    apply_cpu(prev["frequency"], prev["scale"], prev["max_temperature"])
    return {"ok": True, "suggested_scale": best, "mhz": mhz}


def test_gpu(maxmhz, maxmv):
    """Trial a ceiling, bench it, keep it only if vkpeak comes back clean."""
    c = gov_leggi()
    pts = [list(p) for p in c.get("curva") or []]
    maxmhz, maxmv = int(maxmhz), int(maxmv)
    pts = [p for p in pts if p[0] != maxmhz] + [[maxmhz, maxmv]]
    pts.sort()
    for i in range(1, len(pts)):
        if pts[i][1] < pts[i - 1][1]:
            pts[i][1] = pts[i - 1][1]
    cand = dict(c, curva=pts, freq_max=maxmhz)
    r = gov_prova(cand)
    if not r.get("ok"):
        return {"ok": False, "phase": "apply", "err": r.get("err", "applicazione fallita")}
    b = bench_gpu()
    if not b["ok"]:
        gov_annulla()
        return {"ok": False, "applied": False, "bench": b, "err": "Benchmark GPU fallito/instabile. Ripristinato."}
    gov_conferma()
    return {"ok": True, "applied": True, "bench": b}


# ---- the old "get", for the Remote Manager ----------------------------------------
def get():
    out = {"cpu": _cpu_conf(), "gpu": _gpu_compat(), "vram": {"uma_mb": vram_get()}}
    fan = {"mode": "auto", "pct": 0, "rpm": 0}
    try:
        s = json.loads(_rd("/run/skillfish/ventola.json"))
        fan = {"mode": "curve" if s.get("in_controllo") else "auto", "pct": int(s.get("duty") or 0),
               "rpm": next((int(x["value"]) for x in s.get("sensori", []) if x.get("type") == "fan" and x.get("value")), 0)}
    except (OSError, ValueError, TypeError, json.JSONDecodeError):
        # skillfish-fand is not running yet, or the file is mid-write: the
        # default set above is the answer, there is nothing to add here
        pass
    out["fan"] = fan
    cu = {"active": 0, "max": 40, "floor": 7, "rows": {"0.0": 7, "0.1": 7, "1.0": 7, "1.1": 7}, "live": False}
    j = cu_get()
    if j:
        cu.update({"active": j.get("active_cu", 0), "rows": j.get("rows", cu["rows"]),
                   "floor": j.get("floor", 7), "live": True})
    out["cu"] = cu
    return out


# ---- dispatch ---------------------------------------------------------------------
def handle(req):
    c = req.get("cmd")
    if c == "ping":
        return {"ok": True}
    if c == "get":
        return {"ok": True, "data": get()}
    # the governor
    if c == "gov-get":
        return gov_get()
    if c == "gov-set":
        return gov_set(req.get("conf") or {})
    if c == "gov-prova":
        return gov_prova(req.get("conf") or {})
    if c == "gov-conferma":
        return gov_conferma()
    if c == "gov-annulla":
        return gov_annulla()
    if c == "gov-attiva":
        return gov_attiva(bool(req.get("on", True)))
    if c == "apply-gpu":
        return {"ok": apply_gpu_compat(req["minmhz"], req["minmv"], req["maxmhz"], req["maxmv"])}
    if c == "gov-mode":
        return {"ok": True, "mode": "vf"}
    # the CPU
    if c == "apply-cpu":
        return _esito_cpu(*apply_cpu_esito(req["mhz"], req["scale"], req["temp"]))
    if c == "cpu-limits":
        return {"ok": True, "data": cpu_limiti()}
    if c == "persist-cpu":
        ok, perche = apply_cpu_esito(req["mhz"], req["scale"], req["temp"])
        # a setting the backend refused must not be armed for the next boot
        if ok:
            persist_cpu()
        return _esito_cpu(ok, perche)
    if c == "cpu-cores":
        return cpu_cores_get()
    if c == "cpu-cores-set":
        return cpu_cores_set(req.get("cores", []))
    if c == "cpu-smt":
        return cpu_smt_set(bool(req.get("on", True)))
    if c == "cpu-gov":
        return cpu_gov_get()
    if c == "cpu-gov-set":
        return cpu_gov_set(req.get("gov", ""))
    if c == "cpu-pstates-set":
        return cpu_pstates_set(bool(req.get("on", True)))
    if c == "core-unlock":
        return core_unlock_get()
    if c == "core-unlock-set":
        return core_unlock_set(bool(req.get("on", False)))
    if c == "coreunlock-efi":
        return coreunlock_efi_get()
    if c == "coreunlock-efi-set":
        return coreunlock_efi_set(bool(req.get("on", True)))
    if c == "cpu-volatile":
        return cpu_volatile(req["mhz"], req["scale"], req["temp"])
    if c == "thermal-guard":
        return {"ok": thermal_guard(req["limit"])}
    if c == "thermal-guard-get":
        return thermal_guard_get()
    # compute units, memory
    if c == "cu-get":
        return {"ok": True, **cu_get()}
    if c == "cu-apply":
        return cu_apply(req.get("rows", []))
    if c == "cu-keep-boot":
        return cu_keep_boot(req.get("on"), req.get("rows"))
    if c == "cu-test":
        return cu_test()
    if c == "set-vram":
        return {"ok": set_vram(req["mb"]), "reboot": True}
    # scheduler, Mesa
    if c == "scx":
        return scx_get()
    if c == "scx-set":
        return scx_set(bool(req.get("on", False)))
    if c == "scx-azzera":
        return scx_azzera()
    if c == "mesa-sistema":
        return mesa_sistema_get()
    if c == "mesa-sistema-set":
        return mesa_sistema_set(bool(req.get("on", False)))
    # the other helpers
    if c == "kernel":
        return kernel(req.get("azione"), req.get("kv"))
    if c == "snapshot":
        return snapshot(req.get("args", []), int(req.get("attesa", 600)))
    if c == "manutenzione":
        return manutenzione(req.get("args", []))
    if c == "gddr6":
        return gddr6(req.get("azione"))
    if c == "ventola":
        return ventola(req.get("azione"), req.get("dati") or {})
    if c == "servizio":
        return servizio(req.get("azione"), req.get("unit"))
    if c == "gtt":
        return gtt(req.get("mb", 0))
    if c == "unsloth-conf":
        return unsloth_conf()
    if c == "unsloth-conf-set":
        return unsloth_conf_set(bool(req.get("lan")), req.get("parallel", 4))
    if c == "cluster":
        return cluster(req.get("azione", "stato"), req.get("ip", ""),
                       req.get("utente", ""), req.get("password", ""))
    if c == "ai-livello":
        return ai_livello(req.get("livello", ""), req.get("modello", ""))
    if c == "ai-mode":
        return ai_mode(req.get("azione", "stato"))
    if c == "unsloth-password-reset":
        return unsloth_password_reset(req.get("scelta", ""))
    if c == "unsloth-key-set":
        return unsloth_key_set(req.get("key", ""))
    if c == "riavvia":
        _log("riavvio chiesto dal Control Center")
        sh("systemctl reboot")
        return {"ok": True}
    if c == "spegni":
        _log("spegnimento chiesto dal Control Center")
        sh("systemctl poweroff")
        return {"ok": True}
    # benchmarks
    if c == "test-cpu":
        return test_cpu(req["mhz"], req["scale"], req["temp"])
    if c == "test-gpu":
        return test_gpu(req.get("maxmhz"), req.get("maxmv"))
    if c == "suggest-uv":
        return suggest_uv(req["mhz"])
    if c == "bench-cpu":
        return bench_cpu(int(req.get("secs", 60)))
    if c == "bench-gpu":
        return bench_gpu()
    return {"ok": False, "err": "comando sconosciuto"}


def main():
    if len(sys.argv) > 1 and sys.argv[1] == "get":
        print(json.dumps(get()))
        return
    if len(sys.argv) > 1 and sys.argv[1] == "gov-get":
        print(json.dumps(gov_get()))
        return
    sys.stdout.write(json.dumps({"ok": True, "ready": True}) + NL)
    sys.stdout.flush()
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except Exception:
            continue
        if req.get("cmd") == "quit":
            break
        try:
            rep = handle(req)
        except Exception as e:
            rep = {"ok": False, "err": str(e)}
        sys.stdout.write(json.dumps(rep) + NL)
        sys.stdout.flush()


if __name__ == "__main__":
    main()
