#!/usr/bin/env python3
"""gig-log — a light session recorder, so a set can be REVIEWED and not just felt.

The problem it solves: after a run-through, "did it glitch?" and "did it get hot?"
and "which controls did I actually use?" are answered from memory. Memory is a bad
instrument. This writes a 1 Hz timeline to JSONL, wall-clock stamped so it lines up
with the Ardour take, and `--report` renders it back.

WHAT IT RECORDS
  1 Hz sample  package/core temp, fan rpm, cpu freq, throttle counts (as DELTAS),
               per-core cpu%, and per-gear %cpu + RSS for scsynth / sclang /
               ArdourGUI / pulsar (the whole pulsar process group, summed).
  xruns        PipeWire ERR counts per node, also as deltas from a baseline.
  MIDI         every control move on the LCXL (and any other seq port), coalesced.
  markers      free-text annotations (`gig-log.py mark "gimme acid drop"`).

THE LAW: THE OBSERVER MUST NOT PERTURB
Measuring this rig has caused the fault twice now — probe-chain's capture streams
produced the very xruns they were hunting, and the LED daemon's per-event fork made
the lag it was reporting. So:

  * NO audio capture, ever. Nothing here opens a stream.
  * NO per-sample subprocess spawn. Temps, freq, throttle, cpu and RSS all come
    from sysfs/procfs reads. Two long-lived children total (`pw-top -b` for xruns,
    `aseqdump` for MIDI), started once and drained by threads.
  * MIDI IS COALESCED, because a fader sweep is 100+ events/s and a logger that
    writes them all is a firehose that costs more than what it measures. One line
    per (port, channel, controller) per second, carrying count + first/last/min/max.
    That is exactly the #71/#72 fix applied to ourselves. NOTES ARE NEVER DROPPED —
    a note is an event, a CC is a state.
  * Measured cost of the whole thing: see `--selftest`. pw-top alone was 1 jiffy
    per 12 s (0.08% of one core) when this was written.

COUNTERS ARE DELTAS, NOT TOTALS
Two absolute counters on this machine are big and meaningless on their own:
`package_throttle_count` was 17024 after 2 days idle, and Ardour's PipeWire ERR
was 67072. Both count history, including power excursions and device changes that
never touched audio. Only the delta across the session says anything, so the header
records the baseline and every sample records the delta.

USAGE
  tools/gig-log.py record                    # 1 Hz recorder, Ctrl-C to stop
  tools/gig-log.py mark "vague de crime"     # annotate the running session
  tools/gig-log.py report                    # render the newest log
  tools/gig-log.py report FILE.jsonl
  tools/gig-log.py status                    # is a recorder running, what is it writing
  tools/gig-log.py install                   # systemd --user unit (boots with the machine)
  tools/gig-log.py selftest                  # prove the sampler is cheap + parsers work
"""
from __future__ import annotations

import argparse
import glob
import json
import os
import re
import shutil
import signal
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent / "bridge"))
import perf  # noqa: E402  (rootless Thermals/_read live here already — DRY with the Bridge)

LOG_DIR = Path(os.environ.get(
    "GIG_LOG_DIR", os.path.expanduser("~/.local/share/parvagues/gig-log")))

# Gear we care about, by /proc/<pid>/comm. A value of `True` means "sum the whole
# process group" — pulsar is 8 processes (main, GPU, renderers) and the interesting
# number is the total, with the worst single one called out separately.
GEAR = {
    "scsynth":   False,   # the audio server. If this starves, the set dies.
    "sclang":    False,   # SuperDirt / our MIDI bridge
    "ArdourGUI": False,   # recorder + master chain
    "pulsar":    True,    # editor: the known renderer burn (#7/#56)
}

# PipeWire nodes whose xruns actually mean something for us. Anything else is
# still counted into the total, but these get their own line in the report.
XRUN_NODES = ("SuperCollider", "ardour", "Line__sink")

RESCAN_S = 5.0          # how often to re-resolve pids (a restart must not go unnoticed)
PRUNE_DAYS = 30
SPARK = "▁▂▃▄▅▆▇█"


# --------------------------------------------------------------------------- #
# procfs: per-gear cpu% and RSS with no forks
# --------------------------------------------------------------------------- #

def _proc_comm(pid: str) -> str | None:
    return perf._read(f"/proc/{pid}/comm")


def parse_proc_stat(raw: str) -> tuple[int, int] | None:
    """(utime+stime jiffies, rss_pages) from one /proc/<pid>/stat line.

    comm sits in parens and may itself contain spaces AND parens — Chromium-family
    processes like pulsar's renderers do exactly this. Splitting on whitespace from
    the left therefore shifts every field, which is the classic /proc/stat trap. So
    split after the LAST ')'.
    """
    if not raw:
        return None
    try:
        rest = raw[raw.rindex(")") + 2:].split()
        # rest[0] is `state`, i.e. field 3; so field N -> rest[N - 3]
        return int(rest[11]) + int(rest[12]), int(rest[21])   # utime+stime, rss
    except (ValueError, IndexError):
        return None


def _proc_cpu_rss(pid: str) -> tuple[int, int] | None:
    """(utime+stime jiffies, rss_pages) or None if the pid is gone."""
    return parse_proc_stat(perf._read(f"/proc/{pid}/stat") or "")


class GearWatch:
    """Per-gear cpu% + RSS, procfs only.

    Re-resolves pids every RESCAN_S *and* immediately whenever a tracked pid
    vanishes. A binding resolved once at startup and never rechecked is this rig's
    single most common silent failure, and a logger that keeps reporting a dead
    scsynth as "0%" would be worse than no logger.
    """

    def __init__(self, gear: dict[str, bool] = GEAR):
        self.gear = gear
        self._pids: dict[str, list[str]] = {}
        self._prev: dict[str, tuple[float, int]] = {}   # comm -> (mono, jiffies)
        self._last_scan = 0.0
        self._hz = os.sysconf("SC_CLK_TCK") or 100
        self._page_mb = (os.sysconf("SC_PAGE_SIZE") or 4096) / 1024 / 1024
        self.scan()

    def scan(self) -> None:
        found = {c: [] for c in self.gear}
        for p in glob.glob("/proc/[0-9]*/comm"):
            pid = p.split("/")[2]
            c = perf._read(p)
            if c in found:
                found[c].append(pid)
        self._pids = found
        self._last_scan = time.monotonic()

    def sample(self) -> dict[str, dict]:
        now = time.monotonic()
        if now - self._last_scan >= RESCAN_S:
            self.scan()
        out = {}
        for comm, group in self.gear.items():
            pids = self._pids.get(comm) or []
            jif = rss = 0
            worst = 0
            alive = 0
            for pid in pids:
                r = _proc_cpu_rss(pid)
                if r is None:
                    self.scan()          # a pid died: re-resolve NOW, don't wait
                    continue
                alive += 1
                jif += r[0]
                rss += r[1]
                worst = max(worst, r[0])
            if not alive:
                self._prev.pop(comm, None)
                out[comm] = {"up": False}
                continue
            entry = {"up": True, "n": alive, "rss": round(rss * self._page_mb, 1)}
            prev = self._prev.get(comm)
            self._prev[comm] = (now, jif)
            if prev and now > prev[0]:
                dt = now - prev[0]
                entry["cpu"] = round(100 * (jif - prev[1]) / self._hz / dt, 1)
            if group and alive > 1:
                entry["grp"] = alive
            out[comm] = entry
        return out


# --------------------------------------------------------------------------- #
# xruns: one long-lived pw-top, drained by a thread
# --------------------------------------------------------------------------- #

class XrunReader(threading.Thread):
    """Latest per-node PipeWire ERR count, from a single `pw-top -b`.

    pw-top batch mode emits a table ~8x/second. We do NOT want 8 samples/s in the
    log, and we must not let the pipe fill (a full pipe blocks the child, which is
    how the LED daemon's latency grew without bound). So: drain continuously in a
    thread, keep only the newest value per node, and let the 1 Hz loop read it.

    Table layout:  S ID QUANT RATE WAIT BUSY W/Q B/Q ERR [FORMAT...] NAME
    ERR is always token 8; NAME is always the last token (with an optional '+'
    prefix for a linked node). FORMAT is variable-width or empty, which is exactly
    why we index from both ends and never from the middle.
    """

    WARMUP = 5      # sightings per node before its baseline is frozen

    def __init__(self):
        super().__init__(daemon=True)
        self.err: dict[str, int] = {}
        self.base: dict[str, int] = {}
        self._seen: dict[str, int] = {}
        self.lock = threading.Lock()
        self.alive = False
        self.lines = 0
        self._proc: subprocess.Popen | None = None
        self._stop = threading.Event()

    @staticmethod
    def available() -> bool:
        return shutil.which("pw-top") is not None

    def parse_line(self, line: str) -> tuple[str, int] | None:
        f = line.split()
        if len(f) < 10 or f[0] not in ("R", "I", "S", "C", "D", "E", "!"):
            return None
        if not f[1].isdigit():
            return None
        try:
            err = int(f[8])
        except ValueError:
            return None
        name = f[-1].lstrip("+").strip()
        return (name, err) if name else None

    def _baseline(self, name: str, err: int) -> None:
        """Record a value; baseline each node as the MAX over its first few sightings.

        pw-top prints a zero-filled snapshot before the profiler has delivered real
        counters — and it prints more than one of them, non-deterministically. Two
        wrong rules were tried before this one:

          * baseline on FIRST sight  -> captured 0, so Ardour's lifetime 67072 was
            reported as 67072 xruns in a 25-second idle session;
          * baseline on SECOND sight -> usually right, and silently wrong whenever
            pw-top happened to emit two zero tables. A rule that is right most of
            the time is the worst kind for a gig log, because the one bad reading
            looks exactly like a disaster.

        ERR is monotonically non-decreasing within a node's lifetime, so the max
        over a warm-up window IS the true starting count — no timing assumption
        required. WARMUP sightings, not seconds, so a node that appears mid-set
        (he opens something) gets the same treatment.

        A value BELOW the baseline means the node was destroyed and recreated with
        a fresh counter; re-baseline rather than report a negative delta.
        Caller holds the lock.
        """
        n = self._seen.get(name, 0) + 1
        self._seen[name] = n
        self.err[name] = err
        if n <= self.WARMUP:
            self.base[name] = max(self.base.get(name, 0), err)
        elif err < self.base.get(name, 0):
            self.base[name] = err

    def run(self) -> None:
        while not self._stop.is_set():
            try:
                self._proc = subprocess.Popen(
                    ["pw-top", "-b"], stdout=subprocess.PIPE,
                    stderr=subprocess.DEVNULL, text=True, bufsize=1)
            except OSError:
                return
            self.alive = True
            for line in self._proc.stdout:            # blocking read, own thread
                if self._stop.is_set():
                    break
                hit = self.parse_line(line)
                if hit is None:
                    continue
                self.lines += 1
                with self.lock:
                    self._baseline(*hit)
            self.alive = False
            if self._stop.wait(3.0):                 # respawn with backoff
                return

    def deltas(self) -> dict[str, int]:
        with self.lock:
            return {k: v - self.base.get(k, v) for k, v in self.err.items()}

    def total_delta(self) -> int:
        return sum(max(0, v) for v in self.deltas().values())

    def stop(self) -> None:
        self._stop.set()
        if self._proc and self._proc.poll() is None:
            self._proc.terminate()


# --------------------------------------------------------------------------- #
# MIDI: one long-lived aseqdump, coalesced per second
# --------------------------------------------------------------------------- #

CC_RE = re.compile(
    r"^\s*(\d+:\d+)\s+Control change\s+(\d+),\s*controller\s+(\d+),\s*value\s+(\d+)")
NOTE_RE = re.compile(
    r"^\s*(\d+:\d+)\s+Note (on|off)\s+(\d+),\s*note\s+(\d+),\s*velocity\s+(\d+)")
PB_RE = re.compile(r"^\s*(\d+:\d+)\s+Pitch bend\s+(\d+),\s*value\s+(-?\d+)")
PORT_RE = re.compile(r"^\s*(\d+:\d+)\s+(.+?)\s\s+")


def find_seq_port(want: str = "Launch Control XL") -> str | None:
    """Resolve the LCXL's ALSA seq port by NAME, every time we (re)connect.

    Never hardcode it: this device has already moved 20:0 -> 24:0 across a replug,
    and a port number cached at startup is the stale-binding failure again.
    """
    try:
        r = subprocess.run(["aseqdump", "-l"], capture_output=True, text=True,
                           timeout=5)
    except (OSError, subprocess.SubprocessError):
        return None
    for line in (r.stdout or "").splitlines():
        if want.lower() in line.lower() and "HUI" not in line:
            m = re.match(r"\s*(\d+:\d+)", line)
            if m:
                return m.group(1)
    return None


class MidiReader(threading.Thread):
    """Reads one aseqdump and accumulates a COALESCED window.

    Read-only seq subscription — it does not steal MIDI from SuperCollider (SC holds
    the *rawmidi* substream, which is a different resource; that is also why LED
    writes have to go over seq).
    """

    def __init__(self, port: str | None = None, want: str = "Launch Control XL"):
        super().__init__(daemon=True)
        self.want = want
        self.fixed = port
        self.port: str | None = None
        self.cc: dict[tuple[str, int, int], dict] = {}
        self.discrete: list[dict] = []
        self.lock = threading.Lock()
        self.events = 0
        self.alive = False
        self._proc: subprocess.Popen | None = None
        self._stop = threading.Event()

    @staticmethod
    def available() -> bool:
        return shutil.which("aseqdump") is not None

    def feed(self, line: str, now: float) -> bool:
        """Fold one aseqdump line into the window. True if it was an event."""
        m = CC_RE.match(line)
        if m:
            port, ch, cc, val = m.group(1), int(m.group(2)), int(m.group(3)), int(m.group(4))
            key = (port, ch, cc)
            with self.lock:
                e = self.cc.get(key)
                if e is None:
                    self.cc[key] = {"t": now, "n": 1, "v0": val, "v1": val,
                                    "lo": val, "hi": val}
                else:
                    e["n"] += 1
                    e["v1"] = val
                    e["lo"] = min(e["lo"], val)
                    e["hi"] = max(e["hi"], val)
                self.events += 1
            return True
        m = NOTE_RE.match(line)
        if m:
            with self.lock:            # a note is an EVENT: never coalesced away
                self.discrete.append({"t": now, "k": "note", "p": m.group(1),
                                      "on": m.group(2) == "on", "ch": int(m.group(3)),
                                      "note": int(m.group(4)), "vel": int(m.group(5))})
                self.events += 1
            return True
        m = PB_RE.match(line)
        if m:
            key = (m.group(1), int(m.group(2)), -1)
            val = int(m.group(3))
            with self.lock:
                e = self.cc.get(key)
                if e is None:
                    self.cc[key] = {"t": now, "n": 1, "v0": val, "v1": val,
                                    "lo": val, "hi": val}
                else:
                    e["n"] += 1
                    e["v1"] = val
                    e["lo"] = min(e["lo"], val)
                    e["hi"] = max(e["hi"], val)
                self.events += 1
            return True
        return False

    def drain(self) -> tuple[list[dict], list[dict]]:
        """Take the window; returns (coalesced cc records, discrete events)."""
        with self.lock:
            cc, self.cc = self.cc, {}
            disc, self.discrete = self.discrete, []
        out = []
        for (port, ch, num), e in sorted(cc.items()):
            rec = {"t": round(e["t"], 3), "k": "cc" if num >= 0 else "pb",
                   "p": port, "ch": ch, "n": e["n"],
                   "v0": e["v0"], "v1": e["v1"], "lo": e["lo"], "hi": e["hi"]}
            if num >= 0:
                rec["cc"] = num
            out.append(rec)
        return out, disc

    def run(self) -> None:
        while not self._stop.is_set():
            port = self.fixed or find_seq_port(self.want)   # re-resolve every time
            if not port:
                if self._stop.wait(5.0):
                    return
                continue
            self.port = port
            try:
                self._proc = subprocess.Popen(
                    ["aseqdump", "-p", port], stdout=subprocess.PIPE,
                    stderr=subprocess.DEVNULL, text=True, bufsize=1)
            except OSError:
                return
            self.alive = True
            for line in self._proc.stdout:
                if self._stop.is_set():
                    break
                self.feed(line, time.time())
            self.alive = False
            if self._stop.wait(3.0):
                return

    def stop(self) -> None:
        self._stop.set()
        if self._proc and self._proc.poll() is None:
            self._proc.terminate()


# --------------------------------------------------------------------------- #
# the recorder
# --------------------------------------------------------------------------- #

def _uptime_s() -> float:
    raw = perf._read("/proc/uptime") or "0"
    try:
        return float(raw.split()[0])
    except (ValueError, IndexError):
        return 0.0


def _package_throttle_sum() -> int:
    total = 0
    for p in glob.glob(
            "/sys/devices/system/cpu/cpu*/thermal_throttle/package_throttle_count"):
        total += perf._read_int(p, 0) or 0
    return total


class Recorder:
    def __init__(self, out_dir: Path = LOG_DIR, hz: float = 1.0,
                 xruns: bool = True, midi: bool = True,
                 midi_port: str | None = None):
        self.dir = out_dir
        self.period = 1.0 / max(0.05, hz)
        self.th = perf.Thermals()
        self.gear = GearWatch()
        self._pkg_base = _package_throttle_sum()
        self.xr = XrunReader() if (xruns and XrunReader.available()) else None
        self.mid = MidiReader(midi_port) if (midi and MidiReader.available()) else None
        self.path = self._open()
        self.samples = 0
        self._stop = threading.Event()

    def _open(self) -> Path:
        self.dir.mkdir(parents=True, exist_ok=True)
        stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        p = self.dir / f"gig-{stamp}.jsonl"
        self.f = p.open("a", buffering=1)     # line-buffered: a kill -9 keeps the log
        return p

    def prune(self, days: int = PRUNE_DAYS) -> int:
        cut = time.time() - days * 86400
        n = 0
        for p in self.dir.glob("gig-*.jsonl"):
            try:
                if p.stat().st_mtime < cut:
                    p.unlink()
                    n += 1
            except OSError:
                pass
        return n

    def _write(self, rec: dict) -> None:
        self.f.write(json.dumps(rec, separators=(",", ":")) + "\n")

    def header(self) -> dict:
        pl = perf.power_caps_w()
        return {
            "t": round(time.time(), 3), "k": "hdr", "v": 1,
            "iso": datetime.now().isoformat(timespec="seconds"),
            "mono": round(time.monotonic(), 3),
            "uptime": round(_uptime_s()),
            "mode": perf.detect_mode(),
            "gov": perf._read("/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor"),
            "pl1": pl.get("pl1"), "pl2": pl.get("pl2"),
            "fmax": perf._read_int("/sys/devices/system/cpu/cpu0/cpufreq/"
                                   "cpuinfo_max_freq"),
            "thermald": perf.thermald_active(),
            "gpu": perf.gpu_manager_mode(),
            "dgpu": perf.dgpu_state(),
            "base_core_throttle": self.th._throttle_base,
            "base_pkg_throttle": self._pkg_base,
            "period": round(self.period, 3),
            "xruns": bool(self.xr), "midi": bool(self.mid),
            "gear": {c: bool(self.gear._pids.get(c)) for c in GEAR},
        }

    def sample(self) -> dict:
        cores = self.th.cores_c()
        fmax, favg = self.th.freq_mhz()
        pct = self.th.cpu_pct()
        rec = {
            "t": round(time.time(), 3), "k": "s",
            "pkg": self.th.package_c(),
            "core_max": max(cores) if cores else None,
            "fan": self.th.fans_rpm(),
            "fmax": fmax, "favg": favg,
            "cpu_max": max(pct) if pct else None,
            "cpu_avg": round(sum(pct) / len(pct)) if pct else None,
            "thr_core": self.th.throttle_delta(),
            "thr_pkg": _package_throttle_sum() - self._pkg_base,
            "w": self.th.power_draw_w(),
            "gear": self.gear.sample(),
        }
        if self.xr:
            d = self.xr.deltas()
            rec["xrun"] = sum(max(0, v) for v in d.values())
            hot = {k: v for k, v in d.items() if v > 0}
            if hot:
                rec["xrun_by"] = hot
        return rec

    def run(self) -> int:
        for t in (self.xr, self.mid):
            if t:
                t.start()
        self._write(self.header())
        self.prune()
        print(f"gig-log: recording -> {self.path}", file=sys.stderr)
        print(f"gig-log: xruns={'pw-top' if self.xr else 'off'} "
              f"midi={'aseqdump' if self.mid else 'off'} "
              f"period={self.period:.2f}s", file=sys.stderr)
        next_t = time.monotonic()
        try:
            while not self._stop.is_set():
                next_t += self.period
                self._write(self.sample())
                self.samples += 1
                if self.mid:
                    cc, disc = self.mid.drain()
                    for r in disc:
                        r["t"] = round(r["t"], 3)
                        self._write(r)
                    for r in cc:
                        self._write(r)
                self._stop.wait(max(0.0, next_t - time.monotonic()))
        except KeyboardInterrupt:
            pass
        finally:
            self._write({"t": round(time.time(), 3), "k": "end",
                         "samples": self.samples,
                         "midi_events": self.mid.events if self.mid else 0,
                         "xrun": self.xr.total_delta() if self.xr else None})
            for t in (self.xr, self.mid):
                if t:
                    t.stop()
            self.f.close()
        print(f"gig-log: {self.samples} samples -> {self.path}", file=sys.stderr)
        return 0

    def stop(self, *_a) -> None:
        self._stop.set()


# --------------------------------------------------------------------------- #
# marks
# --------------------------------------------------------------------------- #

def newest_log(d: Path = LOG_DIR) -> Path | None:
    logs = sorted(d.glob("gig-*.jsonl"), key=lambda p: p.stat().st_mtime)
    return logs[-1] if logs else None


def cmd_mark(label: str, d: Path = LOG_DIR) -> int:
    """Append a marker to the newest log.

    Deliberately dumb: an O_APPEND write of one short line is atomic enough that
    it can never interleave with the recorder's own lines, so no IPC, no socket,
    nothing that can wedge the recorder. If there is no log, say so and exit 1
    rather than silently creating an orphan.
    """
    p = newest_log(d)
    if p is None:
        print(f"gig-log: no log in {d} — start `gig-log.py record` first",
              file=sys.stderr)
        return 1
    rec = {"t": round(time.time(), 3), "k": "mark", "label": label}
    with p.open("a") as f:
        f.write(json.dumps(rec, separators=(",", ":")) + "\n")
    print(f"mark @ {datetime.now().strftime('%H:%M:%S')}  {label}  -> {p.name}")
    return 0


# --------------------------------------------------------------------------- #
# report
# --------------------------------------------------------------------------- #

def spark(vals: list, width: int = 60) -> str:
    """Sparkline over a fixed width. An aggregate cannot tell a fade from a sparse
    part — keeping the time axis is the whole point (see the probe-chain lesson)."""
    vals = [v for v in vals if v is not None]
    if not vals:
        return "(no data)"
    if len(vals) > width:                     # bucket by MAX: a spike must survive
        step = len(vals) / width
        vals = [max(vals[int(i * step):max(int((i + 1) * step), int(i * step) + 1)])
                for i in range(width)]
    lo, hi = min(vals), max(vals)
    if hi == lo:
        return SPARK[0] * len(vals) + f"  (flat {lo})"
    return "".join(SPARK[min(7, int(8 * (v - lo) / (hi - lo)))] for v in vals)


def gl_warmup() -> int:
    return XrunReader.WARMUP


def hms(seconds: float) -> str:
    s = int(max(0, seconds))
    return f"{s // 3600:d}:{(s % 3600) // 60:02d}:{s % 60:02d}" if s >= 3600 \
        else f"{s // 60:d}:{s % 60:02d}"


# --------------------------------------------------------------------------
# What a PARKED control means
# --------------------------------------------------------------------------
# A move count and a min-max range cannot answer "why is that orbit silent".
# Only the LAST value can, so the report carries it — and for the controls whose
# low end is dangerous, it carries what that value MEANS in Hz.
#
# BootTidal.hs:376
#   gDJF ch = (# lpf (range 180 20000
#                       (fmap (\v -> 1 - 2 * max 0 (0.5 - v)) (orDef 0.5 ch))))
#
# For v <= 0.5 the inner fmap is just `2*v`, so
#       lpf = 180 + 19820 * 2v  =  180 + 39640 * v
# which is LINEAR IN HERTZ. That matters more than it looks: pitch perception is
# logarithmic, so a linear-in-Hz sweep spends nearly all its travel in the top
# two octaves and crosses the entire audible bottom in the last ~2% of the knob.
# The knob therefore feels inert for most of its range and then collapses to
# near-silence right at the end — which is exactly one gesture producing BOTH
# "it went quiet" AND "it sounds lpf'd" (#79).
DJF_CC = (49, 50, 51)          # gF1 gF2 gF3
MASK_CC = 41                   # gMask
MUTE_CC = (73, 74, 75)         # gMute1 gMute2 gMute3
PANIC_CC = 93
ARDOUR_CC = range(77, 85)      # MIDI-learned to Ardour track gains — never seed these

CONTROL_ROLE = {
    **{cc: f"gF{i + 1}  DJ filter (lpf)" for i, cc in enumerate(DJF_CC)},
    MASK_CC: "gMask  someCyclesBy gate",
    **{cc: f"gMute{i + 1}  someCyclesBy mute" for i, cc in enumerate(MUTE_CC)},
    PANIC_CC: "gPanic  global gain kill",
    **{cc: "Ardour track gain" for cc in ARDOUR_CC},
}


def djf_lpf(value: int) -> float:
    """LPF hz that gDJF applies for a raw CC value. See the note above."""
    v = max(0, min(127, int(value))) / 127.0
    x = 1.0 - 2.0 * max(0.0, 0.5 - v)
    return 180.0 + (20000.0 - 180.0) * x


def djf_hpf(value: int) -> float:
    """HPF hz — the half the first version of this report FORGOT.

    gDJF is two-sided (BootTidal.hs:376-377); reading only the `# lpf` line made
    the report call gF3-at-80 "open" when it is really a ~2 kHz high-pass, which
    guts a bass or a voice. Both halves, or the verdict is a lie on one side.
    """
    v = max(0, min(127, int(value))) / 127.0
    return 20.0 + (8000.0 - 20.0) * (2.0 * max(0.0, v - 0.5))


def djf_bands(value: int) -> tuple[float, float]:
    return djf_lpf(value), djf_hpf(value)


def djf_verdict(lpf: float, hpf: float) -> str:
    """How a parked DJ filter will SOUND. Thresholds are by ear, not by maths."""
    if lpf < 400:
        return "⚠⚠ NEAR-SILENT"
    if hpf > 3000:
        return "⚠⚠ NO BODY LEFT"
    if lpf < 1500:
        return "⚠ dark"
    if hpf > 700:
        return "⚠ thin — low end cut"
    if lpf < 6000:
        return "filtered"
    if hpf > 120:
        return "slight hp"
    return "open"


def surface_state(events: list[dict]) -> dict[int, dict]:
    """Fold CC records per control, KEEPING the last value.

    `v1` (the final value inside each coalescing window) is already in the log —
    the report simply never showed it. The last record's `v1` is where the knob
    was left, which is the one number that answers "what is this control doing
    right now".
    """
    agg: dict[int, dict] = {}
    for e in events:
        if e.get("k") != "cc" or "cc" not in e:
            continue
        cc = e["cc"]
        a = agg.get(cc)
        if a is None:
            a = agg[cc] = {"n": 0, "lo": 127, "hi": 0, "first": e["t"],
                           "last": e["t"], "s": 0, "v": e.get("v1"),
                           "v_t": e["t"]}
        a["n"] += e.get("n", 1)
        a["s"] += 1
        a["lo"] = min(a["lo"], e.get("lo", 0))
        a["hi"] = max(a["hi"], e.get("hi", 0))
        a["first"] = min(a["first"], e["t"])
        a["last"] = max(a["last"], e["t"])
        # Guard the ordering rather than trusting file order: a torn or
        # out-of-order tail must not silently become "where the knob is".
        if e["t"] >= a["v_t"]:
            a["v_t"] = e["t"]
            a["v"] = e.get("v1")
    return agg


# Counters that are RUNNING TOTALS. Windowing the session without rebasing these
# reports the whole session's count inside the window — wrong, and wrong HIGH.
CUMULATIVE = ("xrun", "thr_core", "thr_pkg")


def slice_session(samples: list[dict], events: list[dict], marks: list[dict],
                  t_from: float | None, t_to: float | None
                  ) -> tuple[list[dict], list[dict], list[dict]]:
    """Window the session AND rebase its cumulative counters to the window."""
    def win(rows):
        return [r for r in rows
                if (t_from is None or r["t"] >= t_from)
                and (t_to is None or r["t"] <= t_to)]

    sl = win(samples)
    if sl and (t_from is not None or t_to is not None):
        b = sl[0]
        base = {k: (b.get(k) or 0) for k in CUMULATIVE}
        base_by = dict(b.get("xrun_by") or {})
        out = []
        for s in sl:
            r = dict(s)
            for k in CUMULATIVE:
                if s.get(k) is not None:
                    r[k] = max(0, s[k] - base[k])
            if s.get("xrun_by"):
                r["xrun_by"] = {k: max(0, v - base_by.get(k, 0))
                                for k, v in s["xrun_by"].items()}
            out.append(r)
        sl = out
    return sl, win(events), win(marks)


def is_live(samples: list[dict], hdr: dict) -> bool:
    """Is this log still being written RIGHT NOW?

    Decided from the data's own recency, deliberately NOT from a process match:
    `pgrep -f` matches any shell that merely mentions the string, and the unit
    name is easy to get wrong (asking systemd about a mis-guessed
    `parvagues-gig-log` instead of `gig-log.service` returned "inactive" for a
    recorder that was running happily, and that false negative got used to
    resolve the "still running, or killed" ambiguity the wrong way).

    A log whose last sample is within a few periods of now is live. Nothing to
    guess, nothing to name.
    """
    if not samples:
        return False
    period = float(hdr.get("period") or 1.0)
    return (time.time() - samples[-1]["t"]) < max(5.0, period * 5)


def parse_when(spec: str, t0: float) -> float:
    """`+M:SS` / `+H:MM:SS` = offset from session start; `HH:MM[:SS]` = wall clock.

    Wall clock is the useful one, because that is what the Ardour take and PLN's
    memory are both stamped in.
    """
    s = spec.strip()
    if s.startswith("+"):
        off = 0.0
        for part in s[1:].split(":"):
            off = off * 60 + float(part)
        return t0 + off
    parts = [int(x) for x in s.split(":")]
    while len(parts) < 3:
        parts.append(0)
    base = datetime.fromtimestamp(t0)
    want = base.replace(hour=parts[0], minute=parts[1], second=parts[2],
                        microsecond=0)
    ts = want.timestamp()
    return ts + 86400 if ts < t0 else ts     # the session crossed midnight


def load(path: Path) -> tuple[dict, list[dict], list[dict], list[dict], dict]:
    hdr, samples, ccs, marks, end = {}, [], [], [], {}
    with path.open() as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                r = json.loads(line)
            except json.JSONDecodeError:
                continue
            k = r.get("k")
            if k == "hdr":
                hdr = r
            elif k == "s":
                samples.append(r)
            elif k in ("cc", "pb", "note"):
                ccs.append(r)
            elif k == "mark":
                marks.append(r)
            elif k == "end":
                end = r
    return hdr, samples, ccs, marks, end


def cmd_report(path: Path, since: str | None = None,
               until: str | None = None) -> int:
    hdr, samples, events, marks, end = load(path)
    if not samples:
        print(f"gig-log: {path.name} has no samples yet", file=sys.stderr)
        return 1
    t0 = hdr.get("t") or samples[0]["t"]
    live = is_live(samples, hdr)
    win = None
    if since or until:
        # Measure the SET, not the corpus: a recorder that ran all night mixes
        # eight hours of tooling into ninety minutes of playing, and every
        # aggregate then describes the wrong thing.
        t_from = parse_when(since, t0) if since else None
        t_to = parse_when(until, t0) if until else None
        samples, events, marks = slice_session(samples, events, marks,
                                               t_from, t_to)
        if not samples:
            print("gig-log: that window contains no samples", file=sys.stderr)
            return 1
        win = (t_from, t_to)
        t0 = samples[0]["t"]
    dur = samples[-1]["t"] - t0
    W = 74
    p = print
    p("=" * W)
    p(f" GIG LOG  {path.name}")
    p("=" * W)
    p(f" started   {hdr.get('iso', '?')}   (align the Ardour take to this wall clock)")
    if win:
        p(f" WINDOW    {datetime.fromtimestamp(t0):%H:%M:%S}"
          f" → {datetime.fromtimestamp(samples[-1]['t']):%H:%M:%S}"
          f"   (counters rebased to this window)")
    p(f" duration  {hms(dur)}  in {len(samples)} samples "
      f"@ {hdr.get('period', 1)}s")
    p(f" perf mode {hdr.get('mode')}  gov={hdr.get('gov')}  "
      f"PL1={hdr.get('pl1')}W PL2={hdr.get('pl2')}W  "
      f"thermald={'on' if hdr.get('thermald') else 'off'}  gpu={hdr.get('gpu')}")
    gear0 = hdr.get("gear") or {}
    p(f" gear @start " + "  ".join(
        f"{'✓' if v else '✗'}{c}" for c in GEAR for v in [gear0.get(c)]))

    # ---- xruns: the number that decides whether anything else matters --------
    p("")
    p("─ XRUNS " + "─" * (W - 8))
    if not hdr.get("xruns"):
        p(" pw-top was off for this session — no xrun data")
    else:
        xr = [s.get("xrun") for s in samples]
        total = max([v for v in xr if v is not None], default=0)
        p(f" total (delta over the session)  {total}")
        if total:
            p(f" {spark(xr)}")
            by: dict[str, int] = {}
            for s in samples:
                for k, v in (s.get("xrun_by") or {}).items():
                    by[k] = max(by.get(k, 0), v)
            for k, v in sorted(by.items(), key=lambda kv: -kv[1])[:6]:
                p(f"   {v:>6}  {k}")
            first = next((s["t"] - t0 for s in samples if (s.get("xrun") or 0) > 0), None)
            if first is not None:
                p(f" first xrun at +{hms(first)} — check the Ardour take there")
        else:
            p(" ZERO. Whatever else this log says, the audio path held.")
        p(f" (blind spot: each node's first {gl_warmup()} pw-top tables are its"
          f" baseline warm-up, ~1-5 s.")
        p("  Start the recorder before the gear, not after, and this costs nothing.)")

    # ---- thermals -----------------------------------------------------------
    p("")
    p("─ THERMAL " + "─" * (W - 10))
    pk = [s.get("pkg") for s in samples]
    fm = [s.get("fmax") for s in samples]
    known = [v for v in pk if v is not None]
    if known:
        p(f" package  min {min(known)}  mean {round(sum(known)/len(known))}  "
          f"max {max(known)} °C")
        p(f" temp  {spark(pk)}")
    if any(v is not None for v in fm):
        f_ok = [v for v in fm if v is not None]
        p(f" freq  {spark(fm)}   max {max(f_ok)} MHz  mean {round(sum(f_ok)/len(f_ok))} MHz")
    fans = [max(s["fan"]) for s in samples if s.get("fan")]
    if fans:
        p(f" fan   min {min(fans)}  max {max(fans)} rpm")
    thr_c = samples[-1].get("thr_core") or 0
    thr_p = samples[-1].get("thr_pkg") or 0
    p(f" throttle events THIS SESSION   core {thr_c}   package {thr_p}")
    p(f"   (baselines were core {hdr.get('base_core_throttle')} / "
      f"package {hdr.get('base_pkg_throttle')} — absolute counts are meaningless,")
    p(f"    they count power-limit excursions too. Only the delta above is ours.)")
    if thr_p or thr_c:
        p(f" thr   {spark([s.get('thr_pkg') for s in samples])}")
    # the correlation question #8 actually asks
    if hdr.get("xruns"):
        pairs = [(s.get("thr_pkg") or 0, s.get("xrun") or 0) for s in samples]
        d_thr = [b[0] - a[0] for a, b in zip(pairs, pairs[1:])]
        d_xr = [b[1] - a[1] for a, b in zip(pairs, pairs[1:])]
        both = sum(1 for a, b in zip(d_thr, d_xr) if a > 0 and b > 0)
        only_x = sum(1 for a, b in zip(d_thr, d_xr) if a == 0 and b > 0)
        thr_secs = sum(1 for a in d_thr if a > 0)
        if sum(d_xr) == 0:
            p(f" VERDICT  {thr_secs}s of this session throttled and cost ZERO xruns.")
        else:
            p(f" VERDICT  xruns in {both}s that also throttled, "
              f"{only_x}s with xruns and no throttle "
              f"({thr_secs}s throttled in total).")

    # ---- gear ---------------------------------------------------------------
    p("")
    p("─ GEAR " + "─" * (W - 7))
    p(f" {'process':<12} {'cpu% mean':>9} {'max':>6} {'rss mean':>9} {'max':>7}  seen")
    for comm in GEAR:
        cpu = [s["gear"].get(comm, {}).get("cpu") for s in samples]
        cpu = [v for v in cpu if v is not None]
        rss = [s["gear"].get(comm, {}).get("rss") for s in samples]
        rss = [v for v in rss if v is not None]
        up = sum(1 for s in samples if s["gear"].get(comm, {}).get("up"))
        if not up:
            p(f" {comm:<12} {'—':>9} {'—':>6} {'—':>9} {'—':>7}  never")
            continue
        p(f" {comm:<12} {round(sum(cpu)/len(cpu),1) if cpu else 0:>9} "
          f"{max(cpu) if cpu else 0:>6} "
          f"{round(sum(rss)/len(rss)) if rss else 0:>8}M "
          f"{round(max(rss)) if rss else 0:>6}M  {up}/{len(samples)}s")
        if rss and len(rss) > 10 and max(rss) > 1.25 * rss[0]:
            p(f"   ⚠ RSS grew {round(rss[0])}M → {round(max(rss))}M "
              f"({round(100*(max(rss)/rss[0]-1))}%) — the pulsar leak pattern")
        if up < len(samples):
            p(f"   ⚠ MISSING for {len(samples)-up}s of the session (started late, "
              f"or died)")
    cm = [s.get("cpu_max") for s in samples]
    if any(v is not None for v in cm):
        p(f" hottest core  {spark(cm)}  max {max(v for v in cm if v is not None)}%")

    # ---- surface ------------------------------------------------------------
    p("")
    p("─ SURFACE " + "─" * (W - 10))
    ccs = [e for e in events if e.get("k") == "cc"]
    notes = [e for e in events if e.get("k") == "note"]
    if not ccs and not notes:
        p(" no MIDI recorded (controller absent, or aseqdump unavailable)")
    else:
        agg = surface_state(ccs)
        p(f" {len(agg)} controls touched, {sum(a['n'] for a in agg.values())} raw "
          f"events, logged as {len(ccs)} lines "
          f"({sum(a['n'] for a in agg.values()) / max(1,len(ccs)):.0f}x coalescing)")
        p(f" {'cc':>4} {'moves':>6} {'range':>9} {'left at':>7} "
          f"{'first':>7} {'last':>7}  active")
        for cc, a in sorted(agg.items(), key=lambda kv: -kv[1]["n"]):
            flag = "  ⚠ARDOUR-OWNED" if cc in ARDOUR_CC else ""
            if cc in DJF_CC and a["v"] is not None:
                lo, hi = djf_bands(a["v"])
                flag = (f"  lpf {round(lo):>5} hpf {round(hi):>4} Hz  "
                        f"{djf_verdict(lo, hi)}")
            v = "?" if a["v"] is None else str(a["v"])
            p(f" {cc:>4} {a['n']:>6} {a['lo']:>4}-{a['hi']:<4} {v:>7} "
              f"{hms(a['first']-t0):>7} {hms(a['last']-t0):>7} {a['s']:>5}s{flag}")
        if notes:
            p(f" {len(notes)} note events "
              f"({len({n['note'] for n in notes})} distinct notes) — keystep/pads")

    # ---- markers ------------------------------------------------------------
    if marks:
        p("")
        p("─ MARKERS " + "─" * (W - 10))
        for m in marks:
            p(f" +{hms(m['t']-t0):>8}  {m.get('label','')}")
    p("")
    if end:
        p(f" session closed cleanly: {end.get('samples')} samples, "
          f"{end.get('midi_events')} midi events")
    elif live:
        # Say WHICH, rather than offering the reader a choice. The old wording
        # ("still running, or it was killed") made a running recorder look like a
        # possible failure, and that ambiguity got resolved the wrong way once
        # already — costing a whole invented task about a recorder that was fine.
        p(" ● RECORDING NOW — this log is still open, numbers are partial")
    else:
        p(" ⚠ no `end` record and no recent sample: the recorder was KILLED")
        p("   (a clean stop writes one; check `systemctl --user status gig-log`)")
    p("=" * W)
    return 0


# --------------------------------------------------------------------------- #
# status / install / selftest
# --------------------------------------------------------------------------- #

UNIT = """[Unit]
Description=gig-log (ParVagues session recorder: thermals, xruns, gear, MIDI)
Documentation=file://{repo}/tools/gig-log.py

[Service]
Type=simple
WorkingDirectory={repo}
ExecStart={py} {repo}/tools/gig-log.py record
Restart=always
RestartSec=5
Nice=5
# It must never be the reason audio stutters: below every audio thread, and
# capped so a runaway parse cannot eat a core.
CPUWeight=20
IOWeight=20

[Install]
WantedBy=default.target
"""


def cmd_install(enable: bool = True) -> int:
    repo = Path(__file__).resolve().parent.parent
    dest = Path.home() / ".config/systemd/user/gig-log.service"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_text(UNIT.format(repo=repo, py=sys.executable))
    print(f"wrote {dest}")
    steps = [["systemctl", "--user", "daemon-reload"]]
    if enable:
        steps += [["systemctl", "--user", "enable", "--now", "gig-log.service"]]
    for cmd in steps:
        r = subprocess.run(cmd, capture_output=True, text=True)
        print(f"$ {' '.join(cmd)} -> {r.returncode} {(r.stderr or '').strip()}")
        if r.returncode != 0:
            return r.returncode
    print("gig-log: enabled (starts with the session; `loginctl enable-linger` "
          "already on for the Bridge, so it survives logout)")
    return 0


def gmask_is_retired(boot: Path | None = None) -> bool:
    """Is gMask still `midiOn "^41" (mask ...)`, or retired to `id`?

    preflight must judge a control by what it CURRENTLY does, not by what it did
    when the rule was written. ^41 was gMask (a global gate); since 2026-07-29 it
    is d1's gate and gMask is `id`. Same CC, opposite verdict — so read BootTidal
    instead of hardcoding, or this gate goes stale the way every other cached
    binding in this rig has ([[feedback_stale_binding_pattern]]).
    """
    boot = boot or (Path(__file__).resolve().parent.parent / "BootTidal.hs")
    try:
        for raw in boot.read_text().splitlines():
            line = raw.split("--", 1)[0]
            m = re.match(r"\s*gMask\s*=\s*(.+?)\s*$", line)
            if m:
                return m.group(1).strip() == "id"
    except OSError:
        pass
    return False          # unknown => assume it still masks, the cautious read


# Controls whose parked position can only ever be judged by the human: the state
# buttons are gestures PLN arms on purpose. preflight reports them, never fails on
# them -- a gate that cries wolf about intentional performance state is a gate that
# gets ignored, and then it is worth nothing when it matters.
GATE_CC = tuple(list(range(41, 45)) + list(range(57, 61))
                + [76] + list(range(89, 93)))


def cmd_preflight(path: Path) -> int:
    """Is the SURFACE safe to play right now? Exit 0 = yes.

    Why this exists (2026-07-29). Every other check in this toolbox is COLD: pvlint
    parses files, silent-eval queries patterns against an empty control map,
    surface-columns reads .tidal text. All three were green all afternoon while
    gMask sat ARMED AT 127 on the live board for 74 minutes, chopping an eighth out
    of every bar of every orbit. Nothing was broken; a global was simply switched
    on, and no tool in the chain looks at the switches. PLN found it by ear, which
    is the one instrument that should never be the smoke detector.

    So: the missing check is not another parser, it is a reading of LIVE STATE.

    The hard rule here is about a stale log. If the recorder is not running, this
    command must FAIL LOUD rather than report on an old file — a preflight that
    passes because it read yesterday's session is worse than no preflight, and it
    is exactly the failure shape this rig produces over and over.
    """
    hdr, samples, events, marks, end = load(path)
    if not is_live(samples, hdr):
        print("gig-log preflight: FAIL — the recorder is NOT running, so this log "
              "is history, not state.", file=sys.stderr)
        print("  A preflight that reads a stale log is worse than none. Start it:\n"
              "    systemctl --user start gig-log.service", file=sys.stderr)
        return 2

    agg = surface_state(events)
    retired = gmask_is_retired()
    fails: list[str] = []
    warns: list[str] = []
    armed: list[str] = []

    for cc in sorted(agg):
        v = agg[cc]["v"]
        if v is None:
            continue
        if cc == PANIC_CC and v > 0:
            fails.append(f"^{cc} gPanic ARMED at {v} — every gPanic'd stream is "
                         f"gain-killed. This is total silence, by design.")
        elif cc in MUTE_CC and v > 0:
            fails.append(f"^{cc} gMute{MUTE_CC.index(cc) + 1} at {v} — mutes "
                         f"{v / 127.0:.0%} of cycles on every stream it wraps.")
        elif cc in DJF_CC:
            lo, hi = djf_bands(v)
            verdict = djf_verdict(lo, hi)
            line = (f"^{cc} gF{DJF_CC.index(cc) + 1} at {v} — lpf {round(lo)} / "
                    f"hpf {round(hi)} Hz — {verdict}")
            if "⚠⚠" in verdict:
                fails.append(line)
            elif "⚠" in verdict:
                warns.append(line)
        elif cc == MASK_CC and v > 0 and not retired:
            fails.append(f"^{cc} gMask ARMED at {v} — gates "
                         f"{v / 127.0:.0%} of cycles on EVERY gM-wrapped orbit. "
                         f"(Retire it with `gMask = id` and this becomes d1's gate.)")
        elif cc in GATE_CC and v > 0:
            armed.append(f"^{cc} at {v}")
        elif cc in ARDOUR_CC:
            pass          # Ardour's, and MIDI cannot tell us where the strip sits

    W = 74
    print("=" * W)
    print(f" PREFLIGHT  {path.name}   recorder LIVE   "
          f"{len(agg)} control(s) touched this session")
    print(f" gMask: {'retired (^41 is d1 gate)' if retired else 'ACTIVE on ^41'}")
    print("=" * W)
    for line in fails:
        print(f" FAIL  {line}")
    for line in warns:
        print(f" WARN  {line}")
    if armed:
        print(f" armed on purpose (not a failure): {', '.join(armed)}")
    untouched = [cc for cc in DJF_CC + MUTE_CC + (PANIC_CC,) if cc not in agg]
    if untouched:
        print(f" untouched this session, so sitting at the #55 seed (safe): "
              f"{','.join(f'^{c}' for c in untouched)}")
    print("-" * W)
    if fails:
        print(f" NOT SAFE TO PLAY — {len(fails)} blocking, {len(warns)} warning(s).")
        print(" Every one is fixed from the surface: move the knob, unpush the button.")
        return 1
    print(f" SAFE — nothing global is swallowing your sound"
          + (f" ({len(warns)} warning(s) worth a glance)." if warns else "."))
    print(" NOTE this reads the SURFACE only. Ardour faders are not visible over")
    print(" MIDI — run tools/check-mix.py for those, and it reads the SAVED session.")
    return 0


def cmd_controls(path: Path, since: str | None = None,
                 until: str | None = None) -> int:
    """Where is every control PARKED — the 3am question when an orbit is silent.

    Deliberately not folded into `report`: report answers "how did the session
    go", this answers "what is the surface doing right now", and reaching for
    the second while debugging should not mean reading the first.
    """
    hdr, samples, events, marks, end = load(path)
    t0 = hdr.get("t") or (samples[0]["t"] if samples else 0)
    if since or until:
        samples, events, marks = slice_session(
            samples, events, marks,
            parse_when(since, t0) if since else None,
            parse_when(until, t0) if until else None)
    agg = surface_state(events)
    if not agg:
        print("gig-log: no CC recorded in this log/window", file=sys.stderr)
        return 1
    W = 74
    print("=" * W)
    print(f" SURFACE STATE  {path.name}   ({len(agg)} controls touched)")
    print(" the LAST value each control was left at — not its range")
    print("=" * W)
    print(f" {'cc':>4} {'left at':>7} {'at':>8}  {'role':<28} means")
    risky = []
    for cc in sorted(agg):
        a = agg[cc]
        v = a["v"]
        role = CONTROL_ROLE.get(cc, "")
        means = ""
        if cc in DJF_CC and v is not None:
            lo, hi = djf_bands(v)
            verdict = djf_verdict(lo, hi)
            means = f"lpf {round(lo)} / hpf {round(hi)} Hz — {verdict}"
            if "⚠" in verdict:
                risky.append((cc, v, means))
        elif cc in MUTE_CC and v is not None:
            frac = v / 127.0
            means = f"mutes {frac:.0%} of cycles"
            if frac > 0.5:
                risky.append((cc, v, means))
        elif cc == MASK_CC and v is not None:
            means = f"gates {v / 127.0:.0%} of cycles"
        elif cc == PANIC_CC and v is not None:
            means = "PANIC ARMED — global gain 0" if v else "clear"
            if v:
                risky.append((cc, v, means))
        elif cc in ARDOUR_CC:
            means = "Ardour-owned — never seeded from here"
        vs = "?" if v is None else str(v)
        print(f" {cc:>4} {vs:>7} {hms(a['last'] - t0):>8}  {role:<28} {means}")
    print("")
    if risky:
        print(" ⚠ PARKED SOMEWHERE THAT SILENCES OR THINS THE SOUND:")
        for cc, v, means in risky:
            print(f"   cc {cc} = {v}  →  {means}")
        print(" If an orbit is silent and it routes through one of these, look"
              " here FIRST.")
    else:
        print(" No control is parked in a known-dangerous zone.")
    return 0


def cmd_status(d: Path = LOG_DIR) -> int:
    r = subprocess.run(["systemctl", "--user", "is-active", "gig-log.service"],
                       capture_output=True, text=True)
    unit = (r.stdout or "").strip() or "absent"
    # Only RECORDERS count. A concurrent `report`/`mark`/`status` is also a
    # gig-log.py process, and counting those made `status` claim two recorders were
    # running when only one was — the sort of scary-but-wrong reading this whole
    # tool exists to avoid.
    running = []
    for p in glob.glob("/proc/[0-9]*/cmdline"):
        pid = p.split("/")[2]
        if pid == str(os.getpid()):
            continue
        cmd = (perf._read(p) or "").replace("\x00", " ")
        if "gig-log.py" in cmd and " record" in f" {cmd} ":
            running.append(pid)
    print(f"unit gig-log.service : {unit}")
    print(f"processes            : {len(running)} {running}")
    p = newest_log(d)
    if not p:
        print(f"newest log           : none in {d}")
        return 0
    age = time.time() - p.stat().st_mtime
    n = sum(1 for _ in p.open())
    print(f"newest log           : {p}")
    print(f"                       {n} lines, last write {age:.0f}s ago "
          f"({'LIVE' if age < 10 else 'stale'})")
    print(f"all logs             : {len(list(d.glob('gig-*.jsonl')))} files, "
          f"{sum(f.stat().st_size for f in d.glob('gig-*.jsonl'))/1e6:.2f} MB")
    return 0


def cmd_selftest(seconds: float = 6.0) -> int:
    """Prove the two things that would make this tool a liability: that the
    sampler is cheap, and that the parsers actually parse."""
    ok = True
    print("── parsers ──")
    xr = XrunReader()
    cases = [
        ("R  160      0      0  13.6us 462.9us  0.00  0.09  67072                   + ardour",
         ("ardour", 67072)),
        ("C   49      0      0    ---     ---   ---   ---     0                  alsa_output.usb-BEHRINGER_UMC202HD_192k_12345678-00.HiFi__Line__sink",
         ("alsa_output.usb-BEHRINGER_UMC202HD_192k_12345678-00.HiFi__Line__sink", 0)),
        ("S   ID  QUANT   RATE    WAIT    BUSY   W/Q   B/Q  ERR FORMAT           NAME", None),
        ("R  128     512  48000  20.1us  55.0us  0.02  0.05    3 S32LE 2 48000    SuperCollider",
         ("SuperCollider", 3)),
    ]
    for line, want in cases:
        got = xr.parse_line(line)
        flag = "ok " if got == want else "FAIL"
        ok &= got == want
        print(f" {flag} pw-top {str(want):<28} <- {line[:40]!r}")
    md = MidiReader(port="0:0")
    now = 1000.0
    for line in [" 24:0   Control change          0, controller 13, value 64",
                 " 24:0   Control change          0, controller 13, value 90",
                 " 24:0   Note on                 0, note 60, velocity 100",
                 " 24:0   Note on                 0, note 62, velocity 100"]:
        if not md.feed(line, now):
            ok = False
            print(f" FAIL midi did not parse {line!r}")
    cc, disc = md.drain()
    coalesced = len(cc) == 1 and cc[0]["n"] == 2 and cc[0]["lo"] == 64 and cc[0]["hi"] == 90
    print(f" {'ok ' if coalesced else 'FAIL'} 2 CC on one controller -> 1 line "
          f"n=2 range 64-90")
    print(f" {'ok ' if len(disc) == 2 else 'FAIL'} 2 notes -> 2 lines (notes are "
          f"never coalesced)")
    ok &= coalesced and len(disc) == 2

    print(f"\n── cost over {seconds:.0f}s of real sampling ──")
    r = Recorder(out_dir=Path(os.environ.get("TMPDIR", "/tmp")) / "gig-log-selftest",
                 hz=1.0)
    # START the readers. A cost measurement that leaves the drain threads idle is
    # flattering itself — the pw-top pipe is ~8 tables/s and parsing it is most of
    # the work. Measure what actually runs.
    for t in (r.xr, r.mid):
        if t:
            t.start()
    time.sleep(0.5)
    me = str(os.getpid())
    c0 = _proc_cpu_rss(me)
    t0 = time.monotonic()
    for _ in range(int(seconds)):
        r.sample()
        time.sleep(1.0)
    dt = time.monotonic() - t0
    c1 = _proc_cpu_rss(me)
    hz = os.sysconf("SC_CLK_TCK") or 100
    pct = 100 * (c1[0] - c0[0]) / hz / dt
    for t in (r.xr, r.mid):
        if t:
            t.stop()
    r.f.close()
    print(f" sampler + both readers : {pct:.2f}% of one core over {dt:.1f}s")
    print(f" pw-top lines parsed    : {r.xr.lines if r.xr else 0} "
          f"({(r.xr.lines/dt if r.xr else 0):.0f}/s)")
    print(f" rss                    : {c1[1]*4096/1e6:.1f} MB")
    budget = pct < 3.0
    print(f" {'ok ' if budget else 'FAIL'} under the 3%-of-one-core budget "
          f"(the observer must not perturb)")
    ok &= budget
    print(f"\n{'PASS' if ok else 'FAIL'}")
    return 0 if ok else 1


# --------------------------------------------------------------------------- #

def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(
        prog="gig-log.py", description=__doc__.split("\n")[0],
        formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--dir", type=Path, default=LOG_DIR, help="log directory")
    sub = ap.add_subparsers(dest="cmd")

    rec = sub.add_parser("record", help="run the 1 Hz recorder")
    rec.add_argument("--hz", type=float, default=1.0)
    rec.add_argument("--no-xruns", action="store_true",
                     help="skip pw-top (no PipeWire client at all)")
    rec.add_argument("--no-midi", action="store_true", help="skip aseqdump")
    rec.add_argument("--midi-port", help="ALSA seq port (default: find the LCXL)")
    rec.add_argument("--seconds", type=float, help="stop after N seconds")

    mk = sub.add_parser("mark", help="annotate the running session")
    mk.add_argument("label", nargs="+")

    rp = sub.add_parser("report", help="render a log")
    rp.add_argument("file", nargs="?", type=Path)
    rp.add_argument("--from", dest="since", metavar="WHEN",
                    help="window start: HH:MM[:SS] wall clock, or +M:SS from start")
    rp.add_argument("--to", dest="until", metavar="WHEN", help="window end")

    ct = sub.add_parser("controls",
                        help="where every control is PARKED (and what that means)")
    ct.add_argument("file", nargs="?", type=Path)
    ct.add_argument("--from", dest="since", metavar="WHEN")
    ct.add_argument("--to", dest="until", metavar="WHEN")

    pf = sub.add_parser("preflight",
                        help="is the SURFACE safe to play right now? (exit 0 = yes)")
    pf.add_argument("file", nargs="?", type=Path)

    sub.add_parser("status", help="is it running, what is it writing")
    ins = sub.add_parser("install", help="systemd --user unit")
    ins.add_argument("--no-enable", action="store_true")
    st = sub.add_parser("selftest", help="prove the parsers and the cost")
    st.add_argument("--seconds", type=float, default=6.0)

    a = ap.parse_args(argv)
    if a.cmd == "mark":
        return cmd_mark(" ".join(a.label), a.dir)
    if a.cmd in ("report", "controls"):
        p = a.file or newest_log(a.dir)
        if p is None:
            print(f"gig-log: no log found in {a.dir}", file=sys.stderr)
            return 1
        fn = cmd_report if a.cmd == "report" else cmd_controls
        return fn(p, a.since, a.until)
    if a.cmd == "preflight":
        p = a.file or newest_log(a.dir)
        if p is None:
            print(f"gig-log: no log found in {a.dir}", file=sys.stderr)
            return 1
        return cmd_preflight(p)
    if a.cmd == "status":
        return cmd_status(a.dir)
    if a.cmd == "install":
        return cmd_install(enable=not a.no_enable)
    if a.cmd == "selftest":
        return cmd_selftest(a.seconds)
    if a.cmd == "record" or a.cmd is None:
        hz = getattr(a, "hz", 1.0)
        r = Recorder(out_dir=a.dir, hz=hz,
                     xruns=not getattr(a, "no_xruns", False),
                     midi=not getattr(a, "no_midi", False),
                     midi_port=getattr(a, "midi_port", None))
        signal.signal(signal.SIGTERM, r.stop)
        signal.signal(signal.SIGINT, r.stop)
        if getattr(a, "seconds", None):
            threading.Timer(a.seconds, r.stop).start()
        return r.run()
    ap.print_help()
    return 2


if __name__ == "__main__":
    sys.exit(main())
