#!/usr/bin/python3
"""Mood Hyper Engine (root, mood-hyperd.service, runs in its own hyper.slice).

Hyper Cores = 1-2 physical cores (with their SMT siblings) walled off from everything else. The engine
picks the best-binned cores (AMD preferred-core ranking / CPPC highest_perf, else max clock; never the
core holding CPU0, P-cores only on Intel hybrids) and then:

  * pins itself there and runs everything below on those cores,
  * system services: `systemctl set-property --runtime system.slice|init.scope AllowedCPUs=<rest>`,
  * hardware interrupts steered off the Hyper Cores (/proc/irq/*/smp_affinity_list),
  * user processes: sched_setaffinity sweep — lanes:
        focus  = Wayfire, the Mood shell, Xwayland, the focused window's app (whole app scope or
                 process tree) and Hyper jobs may use every core; everything else gets <rest>
        jobs   = only Hyper jobs (`hyper run`) get the Hyper Cores, pinned to them
  * telemetry: RAPL package/core watts (root-only), per-core load, PSI → /run/mood/hyper.json
  * adaptive power (optional): power-profiles-daemon performance/balanced/power-saver from AC + load
  * hardware facts that need root (DIMMs via dmidecode, SMART) → /run/mood/hyper-hw.json

Config /etc/mood/hyper.json {"cores": "auto"|0|1|2, "lane": "focus"|"jobs", "irq": bool, "governor": "manual"|"adaptive"}.
Socket /run/mood/hyper.sock (0666): {"op":"job","name":..} registers the caller (SO_PEERCRED pid) as a
Hyper job and returns the core list; {"op":"status"}. Everything is undone on stop.
"""
import glob
import json
import os
import re
import signal
import socket
import struct
import subprocess
import sys
import threading
import time
from pathlib import Path

CONF = Path("/etc/mood/hyper.json")
RUN = Path(os.environ.get("HYPER_RUN", "/run/mood"))   # override only for tests
STATE = RUN / "hyper.json"
HW = RUN / "hyper-hw.json"
SOCK = str(RUN / "hyper.sock")
CPU = Path("/sys/devices/system/cpu")
DEFAULT = {"cores": "auto", "lane": "focus", "irq": True, "governor": "manual"}
UI_NAMES = {"wayfire", "labwc", "Xwayland", "mood-shell.py", "gtklock", "squeekboard", "mako"}
DRY = os.environ.get("HYPER_DRY") == "1"
VERSION = 1


def log(*a):
    print("hyperd:", *a, flush=True)


def rd(p, d=""):
    try:
        return Path(p).read_text().strip()
    except Exception:
        return d


def rint(p, d=None):
    try:
        return int(rd(p).split()[0])
    except Exception:
        return d


def cpulist(s):
    out = []
    for part in (s or "").split(","):
        part = part.strip()
        if "-" in part:
            a, b = part.split("-", 1)
            out.extend(range(int(a), int(b) + 1))
        elif part:
            out.append(int(part))
    return out


def fmt(cpus):
    cpus = sorted(set(cpus))
    out, i = [], 0
    while i < len(cpus):
        j = i
        while j + 1 < len(cpus) and cpus[j + 1] == cpus[j] + 1:
            j += 1
        out.append(str(cpus[i]) if i == j else f"{cpus[i]}-{cpus[j]}")
        i = j + 1
    return ",".join(out)


def load_conf():
    try:
        c = json.loads(CONF.read_text())
    except Exception:
        c = {}
    out = dict(DEFAULT)
    if c.get("cores") in ("auto", 0, 1, 2):
        out["cores"] = c["cores"]
    if c.get("lane") in ("focus", "jobs"):
        out["lane"] = c["lane"]
    if isinstance(c.get("irq"), bool):
        out["irq"] = c["irq"]
    if c.get("governor") in ("manual", "adaptive"):
        out["governor"] = c["governor"]
    return out


def write_json(path, data, mode=0o644):
    tmp = Path(str(path) + ".tmp")
    tmp.write_text(json.dumps(data))
    os.chmod(tmp, mode)
    tmp.replace(path)


# ---------------------------------------------------------------- core picking
def physical_cores():
    """[(score, sorted logical cpus, kind)] per online physical core."""
    online = set(cpulist(rd(CPU / "online"))) or {int(p.name[3:]) for p in CPU.glob("cpu[0-9]*")}
    p_set = set(cpulist(rd("/sys/devices/cpu_core/cpus")))
    e_set = set(cpulist(rd("/sys/devices/cpu_atom/cpus")))
    groups = {}
    for c in sorted(online):
        sib = tuple(x for x in cpulist(rd(CPU / f"cpu{c}/topology/thread_siblings_list")) if x in online) or (c,)
        groups.setdefault(sib, c)
    out = []
    for sib in groups:
        c = sib[0]
        score = (rint(CPU / f"cpu{c}/cpufreq/amd_pstate_prefcore_ranking")
                 or rint(CPU / f"cpu{c}/cpufreq/amd_pstate_highest_perf")
                 or rint(CPU / f"cpu{c}/acpi_cppc/highest_perf"))
        how = "prefcore" if score else None
        if not score:
            score = rint(CPU / f"cpu{c}/cpufreq/cpuinfo_max_freq")
            how = "maxfreq" if score else "position"
            score = score or 0
        kind = "P" if c in p_set else "E" if c in e_set else ""
        out.append({"cpus": list(sib), "score": score, "how": how, "kind": kind})
    return out, sorted(online)


def pick(want):
    cores, online = physical_cores()
    n_phys = len(cores)
    if want == "auto":
        want = 0 if len(online) < 4 or n_phys < 3 else 1 if n_phys <= 6 else 2
    want = int(want)
    if want <= 0:
        return [], online, None
    cand = [c for c in cores if 0 not in c["cpus"]]
    if any(c["kind"] == "P" for c in cand):
        cand = [c for c in cand if c["kind"] == "P"]
    # best score first; ties → highest cpu number (furthest from cpu0's housekeeping)
    cand.sort(key=lambda c: (c["score"], c["cpus"][0]), reverse=True)
    chosen = []
    for c in cand:
        if len(chosen) >= want:
            break
        rest = [x for x in online if x not in {y for k in chosen + [c] for y in k["cpus"]}]
        if len(rest) < 2:
            break
        chosen.append(c)
    hyper = sorted(x for c in chosen for x in c["cpus"])
    return hyper, online, (chosen[0]["how"] if chosen else None)


# ---------------------------------------------------------------- applying
def systemctl(*a):
    if DRY:
        log("DRY systemctl", *a)
        return True
    try:
        return subprocess.run(["systemctl", *a], capture_output=True, timeout=15).returncode == 0
    except Exception:
        return False


def set_slices(rest):
    val = fmt(rest) if rest else ""
    ok = True
    for unit in ("system.slice", "init.scope"):
        ok &= systemctl("set-property", "--runtime", unit, f"AllowedCPUs={val}")
    if Path("/sys/fs/cgroup/machine.slice").exists():
        systemctl("set-property", "--runtime", "machine.slice", f"AllowedCPUs={val}")
    return ok


IRQ_SAVED = {}


def steer_irqs(hyper, rest, on):
    moved = 0
    for d in glob.glob("/proc/irq/[0-9]*"):
        f = Path(d, "smp_affinity_list")
        try:
            cur = cpulist(f.read_text())
        except Exception:
            continue
        irq = os.path.basename(d)
        if on:
            if hyper and set(cur) & set(hyper):
                new = [c for c in cur if c not in hyper] or rest
                try:
                    if not DRY:
                        f.write_text(fmt(new))
                    IRQ_SAVED.setdefault(irq, cur)
                    moved += 1
                except OSError:
                    pass   # per-CPU / managed IRQs can't move
        elif irq in IRQ_SAVED:
            try:
                if not DRY:
                    f.write_text(fmt(IRQ_SAVED[irq]))
            except OSError:
                pass
    if not on:
        IRQ_SAVED.clear()
    try:
        if on and rest and not DRY:
            Path("/proc/irq/default_smp_affinity").write_text(format(sum(1 << c for c in rest), "x"))
        elif not on and not DRY:
            Path("/proc/irq/default_smp_affinity").write_text(format(sum(1 << c for c in all_cpus()), "x"))
    except OSError:
        pass
    return moved


def all_cpus():
    return cpulist(rd(CPU / "online")) or list(range(os.cpu_count() or 1))


# ---------------------------------------------------------------- processes
def proc_table():
    """pid -> (uid, ppid, comm, cgroup, is_kthread)"""
    out = {}
    for p in os.listdir("/proc"):
        if not p.isdigit():
            continue
        try:
            st = Path(f"/proc/{p}/stat").read_text()
            r = st.rfind(")")
            comm = st[st.find("(") + 1:r]
            f = st[r + 2:].split()
            ppid, flags = int(f[1]), int(f[6])
            uid = os.stat(f"/proc/{p}").st_uid
            out[int(p)] = (uid, ppid, comm, None, bool(flags & 0x00200000))
        except Exception:
            continue
    return out


def cgroup_of(pid):
    try:
        return Path(f"/proc/{pid}/cgroup").read_text().strip().split("::", 1)[-1]
    except Exception:
        return ""


def cmdline_has(pid, word):
    try:
        return word in Path(f"/proc/{pid}/cmdline").read_bytes().decode(errors="replace")
    except Exception:
        return False


def tasks(pid):
    try:
        return [int(t) for t in os.listdir(f"/proc/{pid}/task")]
    except OSError:
        return []


def descendants(roots, table):
    kids = {}
    for pid, (_, ppid, *_r) in table.items():
        kids.setdefault(ppid, []).append(pid)
    seen, stack = set(), list(roots)
    while stack:
        p = stack.pop()
        if p in seen or p not in table:
            continue
        seen.add(p)
        stack.extend(kids.get(p, ()))
    return seen


# ---------------------------------------------------------------- focus (Wayfire IPC)
class Focus:
    def __init__(self):
        self.pid = None
        self.app = ""
        self.title = ""
        self.watching = set()

    def scan(self):
        for path in glob.glob("/tmp/wayfire-*.socket"):
            if path not in self.watching:
                self.watching.add(path)
                threading.Thread(target=self.watch, args=(path,), daemon=True).start()

    @staticmethod
    def _send(s, obj):
        b = json.dumps(obj).encode()
        s.sendall(struct.pack("<I", len(b)) + b)

    @staticmethod
    def _recv(s):
        def need(n):
            buf = b""
            while len(buf) < n:
                c = s.recv(n - len(buf))
                if not c:
                    raise ConnectionError
                buf += c
            return buf
        return json.loads(need(struct.unpack("<I", need(4))[0]))

    def _take(self, v):
        if isinstance(v, dict) and v.get("pid"):
            self.pid, self.app, self.title = int(v["pid"]), v.get("app-id") or "", (v.get("title") or "")[:80]

    def watch(self, path):
        try:
            while os.path.exists(path):
                try:
                    q = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                    q.settimeout(5)
                    q.connect(path)
                    self._send(q, {"method": "window-rules/get-focused-view", "data": {}})
                    self._take((self._recv(q) or {}).get("info"))
                    q.close()
                    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                    s.connect(path)
                    self._send(s, {"method": "window-rules/events/watch", "data": {"events": ["view-focused"]}})
                    self._recv(s)
                    s.settimeout(None)
                    while True:
                        ev = self._recv(s)
                        if ev.get("event") == "view-focused":
                            self._take(ev.get("view"))
                except Exception:
                    time.sleep(3)
        finally:
            self.watching.discard(path)


# ---------------------------------------------------------------- engine
class Engine:
    def __init__(self):
        self.conf = load_conf()
        self.conf_mtime = None
        self.hyper, self.online, self.how = [], all_cpus(), None
        self.rest = list(self.online)
        self.jobs = {}          # pid -> {name, uid, started}
        self.touched = {}       # tid -> mask we set (so we can undo / not fight apps)
        self.focus = Focus()
        self.lock = threading.RLock()
        self.rapl = {}
        self.stat_prev = None
        self.watts = {}
        self.irq_moved = 0
        self.profile = None
        self.gov_hot = 0
        self.self_cpu = (0, time.time())
        self.engine_load = 0.0
        self.core_load = {}
        self.sweeps = 0
        self.boosted = 0
        self.slices_ok = None
        self.bench_lock = threading.Lock()
        self.bench_at, self.bench_last = 0, None

    # ---- config
    def reload(self, force=False):
        try:
            m = CONF.stat().st_mtime
        except FileNotFoundError:
            m = None
        if not force and m == self.conf_mtime:
            return
        self.conf_mtime = m
        old = (self.conf, self.hyper)
        self.conf = load_conf()
        hyper, online, how = pick(self.conf["cores"])
        if force or hyper != self.hyper or self.conf != old[0]:
            self.apply(hyper, online, how)

    def apply(self, hyper, online, how):
        with self.lock:
            self.hyper, self.online, self.how = hyper, online, how
            self.rest = [c for c in online if c not in hyper] or online
            log(f"hyper cores {fmt(hyper) or 'off'} (picked by {how}), rest {fmt(self.rest)}, lane {self.conf['lane']}")
            self.slices_ok = set_slices(self.rest if hyper else [])
            steer_irqs(self.hyper, self.rest, False)
            self.irq_moved = steer_irqs(self.hyper, self.rest, True) if hyper and self.conf["irq"] else 0
            try:
                os.sched_setaffinity(0, set(hyper) if hyper else set(online))
            except OSError:
                pass
            self.touched.clear()
            self.sweep(full=True)

    def release(self):
        log("releasing hyper cores")
        with self.lock:
            set_slices([])
            steer_irqs(self.hyper, self.rest, False)
            everything = set(all_cpus())
            for tid, mask in list(self.touched.items()):
                try:
                    if set(os.sched_getaffinity(tid)) == set(mask):
                        os.sched_setaffinity(tid, everything)
                except OSError:
                    pass
            self.touched.clear()
        try:
            STATE.unlink()
        except OSError:
            pass

    # ---- the sweep
    def classify(self, table):
        boost, jobs = set(), set()
        for pid in list(self.jobs):
            if pid not in table:
                self.jobs.pop(pid, None)
        jobs = descendants(self.jobs.keys(), table)
        if self.conf["lane"] == "focus":
            ui, roots = set(), []
            for pid, (uid, _pp, comm, _cg, kt) in table.items():
                if kt or uid < 1000:
                    continue
                if comm in UI_NAMES or (comm.startswith("python") and cmdline_has(pid, "mood-shell.py")):
                    ui.add(pid)   # the UI process itself, not everything it ever spawned
            fp = self.focus.pid
            if fp and fp in table:
                cg = cgroup_of(fp)
                if re.search(r"/app\.slice/(app-|flatpak-|snap\.)", cg):
                    for pid, row in table.items():
                        if row[0] == table[fp][0] and cgroup_of(pid) == cg:
                            roots.append(pid)
                else:
                    roots.append(fp)
            boost = ui | descendants(roots, table)
        return boost, jobs

    def sweep(self, full=False):
        if not self.hyper:
            return
        table = proc_table()
        boost, jobs = self.classify(table)
        H, R, A = set(self.hyper), set(self.rest), set(self.online)
        n_boost = 0
        for pid, (uid, _pp, _c, _cg, kt) in table.items():
            if kt or uid < 1000:
                continue
            want = H if pid in jobs else A if pid in boost else R
            if pid in boost:
                n_boost += 1
            for tid in tasks(pid):
                if not full and self.touched.get(tid) == want:
                    continue
                try:
                    cur = set(os.sched_getaffinity(tid))
                except OSError:
                    continue
                if want is R:
                    # keep narrower masks an app chose itself, just without the hyper cores
                    new = (cur & R) or R
                elif want is A:
                    mine = self.touched.get(tid)
                    new = A if (mine is not None and cur == set(mine)) or cur == R or cur == H else cur
                else:
                    new = H
                if new != cur:
                    try:
                        if not DRY:
                            os.sched_setaffinity(tid, new)
                    except OSError:
                        continue
                self.touched[tid] = new
        if full:
            live = {t for p in table for t in tasks(p)}
            for tid in list(self.touched):
                if tid not in live:
                    self.touched.pop(tid, None)
        self.boosted = n_boost
        self.sweeps += 1

    # ---- telemetry
    def sample(self):
        now = time.time()
        for z in glob.glob("/sys/class/powercap/*rapl*:*"):
            name = rd(f"{z}/name")
            e = rint(f"{z}/energy_uj")
            if e is None or not name:
                continue
            rng = rint(f"{z}/max_energy_range_uj", 1 << 32)
            prev = self.rapl.get(z)
            if prev:
                de = e - prev[0]
                if de < 0:
                    de += rng
                dt = now - prev[1]
                if dt > 0:
                    key = "package" if name.startswith("package") else name
                    self.watts[key] = round(de / dt / 1e6, 2)
            self.rapl[z] = (e, now)
        times = {}
        for line in rd("/proc/stat").splitlines():
            if line.startswith("cpu") and line[3:4].isdigit():
                f = line.split()
                v = list(map(int, f[1:9]))
                times[int(f[0][3:])] = (sum(v), v[3] + v[4])
        if self.stat_prev:
            for c, (tot, idle) in times.items():
                p = self.stat_prev.get(c)
                if p and tot > p[0]:
                    self.core_load[c] = round(100 * (1 - (idle - p[1]) / (tot - p[0])), 1)
        self.stat_prev = times
        t = os.times()
        cpu_s = t.user + t.system
        ps, pt = self.self_cpu
        self.engine_load = round(100 * (cpu_s - ps) / max(0.001, now - pt), 2)
        self.self_cpu = (cpu_s, now)

    # ---- adaptive power
    def governor(self):
        if self.conf["governor"] != "adaptive":
            return
        ac, cap = True, None
        for p in Path("/sys/class/power_supply").glob("*"):
            t = rd(p / "type")
            if t == "Battery" and rd(p / "scope") != "Device":
                ac = ac and rd(p / "status") in ("Charging", "Full", "Not charging")
                cap = rint(p / "capacity")
            elif t in ("Mains", "USB", "USB_C", "USB_PD") and rd(p / "online") == "1":
                ac = True
        busy = sum(self.core_load.get(c, 0) for c in self.rest) / max(1, len(self.rest))
        self.gov_hot = self.gov_hot + 1 if busy > 45 else 0
        if cap is not None and not ac and cap <= 20:
            want = "power-saver"
        elif ac and self.gov_hot >= 2:
            want = "performance"
        elif not ac and busy < 8:
            want = "power-saver"
        else:
            want = "balanced"
        if want != self.profile:
            if not DRY and subprocess.run(["powerprofilesctl", "set", want], capture_output=True, timeout=10).returncode != 0 and want == "performance":
                want = "balanced"
                subprocess.run(["powerprofilesctl", "set", want], capture_output=True, timeout=10)
            self.profile = want
            log("power profile →", want)

    def publish(self):
        jobs = [{"pid": p, **j} for p, j in self.jobs.items()]
        st = {"version": VERSION, "at": time.time(), "running": True, "config": self.conf,
              "cores": self.hyper, "rest": self.rest, "online": self.online, "pickedBy": self.how,
              "lane": self.conf["lane"], "focus": {"pid": self.focus.pid, "app": self.focus.app, "title": self.focus.title, "ipc": bool(self.focus.watching)},
              "boosted": self.boosted, "jobs": jobs, "watts": self.watts, "coreLoad": {str(c): self.core_load.get(c) for c in self.hyper},
              "engineLoad": self.engine_load, "irqMoved": self.irq_moved, "slices": self.slices_ok, "profile": self.profile,
              "rapl": bool(self.rapl), "sweeps": self.sweeps}
        write_json(STATE, st)

    # ---- root-only hardware facts
    def hw_scan(self):
        out = {"at": time.time(), "dimms": [], "smart": {}}
        try:
            txt = subprocess.run(["dmidecode", "-t", "16,17"], capture_output=True, text=True, timeout=20).stdout
            for blk in txt.split("\n\n"):
                kv = dict(re.findall(r"^\t([^:\n]+):\s*(.*)$", blk, re.M))
                if "Memory Device" in blk:
                    if kv.get("Size", "No Module Installed").startswith(("No", "Unknown")):
                        continue
                    out["dimms"].append({"slot": kv.get("Locator"), "bank": kv.get("Bank Locator"), "size": kv.get("Size"), "type": kv.get("Type"),
                                         "form": kv.get("Form Factor"), "speed": (kv.get("Speed") or "").replace(" MT/s", "").replace(" MHz", ""),
                                         "configured": (kv.get("Configured Memory Speed") or kv.get("Configured Clock Speed") or "").replace(" MT/s", "").replace(" MHz", ""),
                                         "maker": kv.get("Manufacturer"), "part": (kv.get("Part Number") or "").strip(), "rank": kv.get("Rank"),
                                         "voltage": kv.get("Configured Voltage"), "width": kv.get("Data Width")})
                elif "Physical Memory Array" in blk:
                    out["maxCapacity"] = kv.get("Maximum Capacity")
                    out["slots"] = kv.get("Number Of Devices")
        except Exception as e:
            out["dmiError"] = str(e)
        if os.path.exists("/usr/sbin/smartctl"):
            for b in Path("/sys/block").iterdir():
                if not re.match(r"(nvme\d+n\d+|sd[a-z]+)$", b.name) or rd(b / "removable") == "1":
                    continue
                try:
                    j = json.loads(subprocess.run(["/usr/sbin/smartctl", "-j", "-a", f"/dev/{b.name}"], capture_output=True, text=True, timeout=25).stdout or "{}")
                except Exception:
                    continue
                nv = j.get("nvme_smart_health_information_log") or {}
                ata = {a.get("name"): a.get("raw", {}).get("value") for a in (j.get("ata_smart_attributes") or {}).get("table", [])}
                out["smart"][b.name] = {
                    "passed": (j.get("smart_status") or {}).get("passed"), "temp": (j.get("temperature") or {}).get("current"),
                    "hours": (j.get("power_on_time") or {}).get("hours"), "cycles": j.get("power_cycle_count"),
                    "used": nv.get("percentage_used"), "spare": nv.get("available_spare"),
                    "writtenTB": round(nv["data_units_written"] * 512000 / 1e12, 2) if nv.get("data_units_written") else
                                 (round(ata["Total_LBAs_Written"] * 512 / 1e12, 2) if ata.get("Total_LBAs_Written") else None),
                    "readTB": round(nv["data_units_read"] * 512000 / 1e12, 2) if nv.get("data_units_read") else None,
                    "unsafeShutdowns": nv.get("unsafe_shutdowns"), "mediaErrors": nv.get("media_errors"),
                    "realloc": ata.get("Reallocated_Sector_Ct"), "family": j.get("model_family")}
        write_json(HW, out)

    # ---- core sprint: single-thread speed of every physical core (thread pinned per core)
    def bench(self):
        if not self.bench_lock.acquire(blocking=False):
            raise ValueError("a core sprint is already running")
        try:
            if time.time() - self.bench_at < 20 and self.bench_last:
                return self.bench_last
            cores, _ = physical_cores()
            cores = cores[:32]
            out = []
            for c in cores:
                cpu = c["cpus"][0]
                try:
                    os.sched_setaffinity(0, {cpu})
                except OSError:
                    continue
                x, n, end = 1, 0, time.perf_counter() + 0.3
                while time.perf_counter() < end:
                    for _ in range(2000):
                        x = (x * 1103515245 + 12345) & 0x7FFFFFFF
                    n += 1
                out.append({"cpus": c["cpus"], "raw": n, "hyper": cpu in self.hyper})
            try:
                os.sched_setaffinity(0, set(self.hyper) if self.hyper else set(self.online))
            except OSError:
                pass
            top = max((o["raw"] for o in out), default=1) or 1
            for o in out:
                o["score"] = round(o["raw"] / top * 1000)
            self.bench_last, self.bench_at = {"at": time.time(), "cores": out}, time.time()
            return self.bench_last
        finally:
            self.bench_lock.release()

    # ---- socket
    def serve(self):
        try:
            os.unlink(SOCK)
        except FileNotFoundError:
            pass
        srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        srv.bind(SOCK)
        os.chmod(SOCK, 0o666)
        srv.listen(16)
        while True:
            c, _ = srv.accept()
            threading.Thread(target=self.client, args=(c,), daemon=True).start()

    def client(self, c):
        try:
            c.settimeout(5)
            pid, uid, _gid = struct.unpack("3i", c.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
            req = json.loads(c.makefile().readline() or "{}")
            op = req.get("op")
            if op == "job":
                mine = [p for p, j in self.jobs.items() if j["uid"] == uid]
                if len(mine) >= 8:
                    raise ValueError("too many Hyper jobs running")
                name = re.sub(r"[^\w .+:/@-]", "", str(req.get("name") or "job"))[:60]
                with self.lock:
                    self.jobs[pid] = {"name": name, "uid": uid, "started": time.time()}
                    if self.hyper and not DRY:
                        try:
                            os.sched_setaffinity(pid, set(self.hyper))
                            self.touched[pid] = set(self.hyper)
                        except OSError:
                            pass
                reply = {"ok": True, "cores": self.hyper}
            elif op == "bench":
                reply = {"ok": True, "result": self.bench()}
            elif op == "status":
                reply = {"ok": True, "state": json.loads(STATE.read_text()) if STATE.exists() else None}
            else:
                reply = {"ok": False, "error": "unknown op"}
        except Exception as e:
            reply = {"ok": False, "error": str(e)}
        try:
            c.sendall((json.dumps(reply) + "\n").encode())
        except OSError:
            pass
        c.close()

    # ---- main loop
    def run(self):
        RUN.mkdir(parents=True, exist_ok=True)
        threading.Thread(target=self.serve, daemon=True).start()
        threading.Thread(target=self.hw_loop, daemon=True).start()
        try:
            self.reload(force=True)
        except Exception as e:
            log("first apply failed:", repr(e))
        tick = 0
        while True:
            t0 = time.time()
            try:
                if tick % 3 == 0:
                    self.reload()
                    self.focus.scan()
                with self.lock:
                    self.sweep(full=tick % 10 == 0)
                self.sample()
                if tick % 5 == 0:
                    self.governor()
                self.publish()
            except Exception as e:
                log("tick error:", repr(e))
            tick += 1
            time.sleep(max(0.2, 1.0 - (time.time() - t0)))

    def hw_loop(self):
        while True:
            try:
                self.hw_scan()
            except Exception as e:
                log("hw scan:", e)
            time.sleep(600)


def main():
    if os.geteuid() != 0 and not DRY and "HYPER_RUN" not in os.environ:
        sys.exit("mood-hyperd must run as root")
    if "--pick" in sys.argv:
        hyper, online, how = pick(load_conf()["cores"] if len(sys.argv) < 3 else (sys.argv[2] if sys.argv[2] == "auto" else int(sys.argv[2])))
        print(json.dumps({"hyper": hyper, "rest": [c for c in online if c not in hyper], "how": how}))
        return
    eng = Engine()

    def stop(*_):
        eng.release()
        os._exit(0)
    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGHUP, lambda *_: threading.Thread(target=eng.reload, kwargs={"force": True}, daemon=True).start())
    eng.run()


if __name__ == "__main__":
    main()
