#!/usr/bin/env python3
"""istrain dashboard server (LOCAL preview).

Serves the static dashboard AND a tiny control endpoint for the per-radio NOAA sanity toggle.
Replaces `python3 -m http.server` so a browser button can start/stop live audio per dongle.

  /api/noaa   GET            -> {"playing": bool, "serial": <playing serial|null>, "freq", "channel"}
              POST ?serial=X  -> toggle dongle X on NOAA: if X is playing, stop it; otherwise stop
                                 whatever's playing and tune X. One dongle plays at a time.

  BUILD vs OPERATE (radio-runtime rule): the maintainer builds this server; the USER OPERATES the radio by
  clicking a NOAA button in the browser. The server spawns rtl_fm ONLY in response to a POST driven
  by that click. the code author must NOT curl/POST this endpoint to start audio — that's operating. GET is
  read-only (no radio op) and safe to hit.

  It's a *temporary hijack* to confirm a radio is responding: while NOAA plays, that dongle is busy;
  stopping frees it so it can go back to its assigned task. NOAA voice = signal; plain hiss =
  receiver works but weak/no signal (squelch is off so the test always makes sound).
"""
import json, os, glob, re, signal, subprocess, atexit, shutil, time, threading, hashlib
from urllib.parse import urlparse, parse_qs
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer

HERE   = os.path.dirname(os.path.abspath(__file__))
PORT   = int(os.environ.get("PORT", "8157"))
BIND   = os.environ.get("BIND", "127.0.0.1")

# NOAA sanity-toggle config (env-overridable; later this comes from the swappable roles file).
NOAA_FREQ    = os.environ.get("NOAA_FREQ", "162.550M")        # Columbus KIG76
NOAA_CHANNEL = os.environ.get("NOAA_CHANNEL", "162.550 · Columbus KIG76")
NOAA_GAIN    = os.environ.get("NOAA_GAIN", "auto")            # "auto" or a dB number
NOAA_DEFAULT = os.environ.get("NOAA_SERIAL", "1001")          # used if POST omits ?serial

# Airband captures: list + serve the per-transmission MP3 clips. Reads the dir passively — it does
# NOT care how Airband was started (run by hand, or the istrain-airband service). See airband/istrain.conf.
CAPTURES_DIR = os.environ.get("CAPTURES_DIR", "/tmp/airband")
# Persistent "heard" log — one line per transmission ever detected (channel + time + size). Survives
# clip pruning AND reboot, so it's the durable substrate for spotting train patterns. NOT in /tmp.
HEARD_LOG = os.path.expanduser(os.environ.get("HEARD_LOG", "~/istrain/heard.jsonl"))
EOT_LOG = os.path.expanduser(os.environ.get("EOT_LOG", "~/istrain/eot.jsonl"))
BOT_LOG = os.path.expanduser(os.environ.get("BOT_LOG", "~/istrain/bot.jsonl"))         # 452 head-end (when the hop runs)
MID_LOG = os.path.expanduser(os.environ.get("MID_LOG", "~/istrain/midtrain.jsonl"))    # DPU 452/457 ±12.5k (when IQ sub-channel ID feeds it; NOT auto from bot/eot)
LEVEL_LOG = os.path.expanduser(os.environ.get("LEVEL_LOG", "~/istrain/level.json"))    # vumon's live instantaneous level (meter)
# Supervisor snapshot: on the NAS (presentation node) /api/supervisor serves a snapshot PUSHED from the
# VM (the NAS has no systemd/dongles). Set SUPERVISOR_SNAPSHOT on the NAS; empty on the VM = serve live.
SUPERVISOR_SNAPSHOT = os.path.expanduser(os.environ.get("SUPERVISOR_SNAPSHOT", ""))
# Push receiver (NAS): the VM POSTs its data files here over Tailscale (bypasses Cloudflare). PUSH_KEY
# gates it — if EMPTY, /api/push is DISABLED (so the VM's own serve.py, which has no key, can't be pushed
# to). Each whitelisted name maps to the exact path its read endpoint reads, so a push = "replace that file".
PUSH_KEY = os.environ.get("PUSH_KEY", "")
SITE_WINDOW_S = int(os.environ.get("SITE_WINDOW_DAYS", "3")) * 86400   # website shows the last N days only; the raw logs keep EVERYTHING (for the deeper offline analysis)
# Errors card (2026-07-06): ERRORS_LOG = this node's own events (ctl actions, failures, client JS
# reports). ERRORS_PUSHED = the VM's pushed copy (NAS only) — defaults NEXT TO the supervisor snapshot
# so the existing compose needs no new env. /api/errors merges both.
ERRORS_LOG = os.path.expanduser(os.environ.get("ERRORS_LOG", "~/istrain/errors.jsonl"))
ERRORS_PUSHED = os.environ.get("ERRORS_PUSHED", "") or (
    os.path.join(os.path.dirname(SUPERVISOR_SNAPSHOT), "errors-vm.jsonl") if SUPERVISOR_SNAPSHOT else "")
# Voice transcripts (2026-07-10): real-speech lines from the Mill's whisper (transcribe-worker).
TRANSCRIPTS_LOG = os.path.expanduser(os.environ.get("TRANSCRIPTS_LOG", "~/istrain/transcripts.jsonl"))
_PUSH_TARGETS = {
    "eot": EOT_LOG, "bot": BOT_LOG, "midtrain": MID_LOG, "heard": HEARD_LOG,
    "level": LEVEL_LOG, "supervisor": SUPERVISOR_SNAPSHOT or os.path.expanduser("~/istrain/supervisor.json"),
    "errors": ERRORS_PUSHED or os.path.expanduser("~/istrain/errors-vm.jsonl"),
    "transcripts": TRANSCRIPTS_LOG,
}
# Hit counter: total page loads + unique visitors. Counted on a real page load (GET /), never on the
# API polls. Behind Cloudflare the visitor IP is in CF-Connecting-IP; IPs are stored HASHED (dedup +
# privacy). Persist to /data on the NAS (set HITS_LOG) so counts survive container restarts.
HITS_LOG = os.path.expanduser(os.environ.get("HITS_LOG", "~/istrain/hits.json"))
_hits_lock = threading.Lock()
_CH_LABELS = {
    "ns_dayton_road_160980":  "NS Dayton Rd · 160.980 ★",
    "ns_buckeye_yard_160920": "NS Buckeye Yard · 160.920",
    "ns_col_sandusky_161190": "NS Col+Sandusky · 161.190",
    "csx_scottslawn_160860":  "CSX Scottslawn · 160.860",
    "ns_col_area_161085":     "NS Col area · 161.085",
}

_proc   = None   # the running rtl_fm|aplay pipeline (Popen), or None
_serial = None   # which dongle serial it's tuned to, or None

# ── jsonl parse cache ────────────────────────────────────────────────────────────────────────────
# The dashboard polls every 5 s but the logs only change every ~20-40 s (append) or on a decode
# enrich (atomic rewrite) — yet every poll was re-parsing ~2 MB / ~40k lines of jsonl (profiled at
# ~100 ms per /api/trains call, 2026-07-05). Cache the parsed rows keyed on (mtime_ns, size); any
# append or rewrite changes the key. Callers get the SAME list object on a hit — treat it as
# READ-ONLY: filter/sort into new lists, never .sort()/.append() the returned one.
_jsonl_cache = {}
_jsonl_lock = threading.Lock()


def _read_jsonl(path):
    try:
        st = os.stat(path)
    except OSError:
        return []
    key = (st.st_mtime_ns, st.st_size)
    with _jsonl_lock:
        hit = _jsonl_cache.get(path)
        if hit and hit[0] == key:
            return hit[1]
    rows = []
    try:
        with open(path) as f:
            for line in f:
                line = line.strip()
                if not line: continue
                try: rows.append(json.loads(line))
                except Exception: pass
    except OSError:
        pass
    with _jsonl_lock:
        _jsonl_cache[path] = (key, rows)
    return rows


def _dongle_state(serial):
    """(present, dvb_claimed) for a dongle by serial — passive sysfs read, never opens it."""
    for d in glob.glob("/sys/bus/usb/devices/*"):
        try:
            if open(os.path.join(d, "idVendor")).read().strip() != "0bda": continue
            if open(os.path.join(d, "idProduct")).read().strip() != "2838": continue
            if open(os.path.join(d, "serial")).read().strip() != serial: continue
        except OSError:
            continue
        for intf in glob.glob(d + ":*"):
            link = os.path.join(intf, "driver")
            if os.path.islink(link) and os.path.basename(os.readlink(link)) == "dvb_usb_rtl28xxu":
                return True, True
        return True, False
    return False, False


def _playing():
    global _proc, _serial
    if _proc is not None and _proc.poll() is not None:
        _proc, _serial = None, None   # it exited on its own (dongle busy, killed, etc.)
    return _proc is not None


def _stop():
    """Kill the whole pipeline group (rtl_fm AND aplay), freeing the dongle."""
    global _proc, _serial
    if _proc is not None:
        try:
            os.killpg(os.getpgid(_proc.pid), signal.SIGTERM)
        except (ProcessLookupError, PermissionError):
            pass
        try:
            _proc.wait(timeout=3)
        except subprocess.TimeoutExpired:
            try: os.killpg(os.getpgid(_proc.pid), signal.SIGKILL)
            except ProcessLookupError: pass
    _proc, _serial = None, None


def _start(serial):
    """Tune NOAA on `serial` and play it. Returns None on success or an error string."""
    global _proc, _serial
    if not shutil.which("rtl_fm"):
        return "rtl_fm not found — install the 'rtl-sdr' package."
    present, dvb = _dongle_state(serial)
    if not present:
        return f"dongle {serial} not present — check the radios panel."
    if dvb:
        return f"dongle {serial} is DVB-claimed — free it: sudo modprobe -r rtl2832_sdr dvb_usb_rtl28xxu"
    # squelch off (-l 0) so the test ALWAYS makes sound: NOAA voice = signal, hiss = receiver works.
    gain = "" if NOAA_GAIN == "auto" else f"-g {NOAA_GAIN} "
    cmd = (f"rtl_fm -d {serial} -f {NOAA_FREQ} -M fm -s 200000 -r 48000 -E deemp -l 0 {gain}- "
           f"| aplay -r 48000 -f S16_LE -t raw -c 1")
    # own session so we can kill the pair as a group; detached from the request handler thread.
    _proc = subprocess.Popen(cmd, shell=True, start_new_session=True,
                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    _serial = serial
    return None


def _state_json(error=None):
    d = {"playing": _playing(), "serial": _serial if _playing() else None,
         "freq": NOAA_FREQ, "channel": NOAA_CHANNEL}
    if error: d["error"] = error
    return d


def _toggle(serial):
    """One-at-a-time: same dongle -> stop; different (or idle) -> stop any, start this one."""
    if _playing() and _serial == serial:
        _stop()
        return None
    _stop()
    return _start(serial)


def _capture_label(fname):
    """Friendly channel label from a recording filename (<template>_<timestamp...>.mp3).
    Prefix match on the known templates — robust to whatever date/time suffix Airband appends
    (hourly buckets vs split_on_transmission both work)."""
    for key, label in _CH_LABELS.items():
        if fname.startswith(key):
            return label
    m = re.findall(r"\d{6}", fname)                      # fallback: first 6-digit group = the freq
    return f"{m[0][:3]}.{m[0][3:]} MHz" if m else re.sub(r"\.mp3$", "", fname)


def _captures_json():
    """Newest-first list of finalized .mp3 clips in CAPTURES_DIR (skips in-progress .tmp).
    Each clip carries its whisper transcript when a real-speech sidecar .txt exists (2026-07-10).
    "writable" tells the panel whether delete is allowed — VM yes; the public NAS node no
    (PUSH_KEY set = presentation node = read-only clips)."""
    # size floor (the owner 2026-07-11): the squelch taps ~47k clips/day, median 2 KB (<1 s of audio,
    # unintelligible) — they bury the real transmissions in minutes. LIST only clips big enough to
    # be speech; every file stays on disk (trim owns cleanup), this is presentation only.
    min_bytes = int(os.environ.get("CAPTURE_MIN_KB", "8")) * 1024
    out = []
    for p in glob.glob(os.path.join(CAPTURES_DIR, "*.mp3")):
        try:
            st = os.stat(p)
        except OSError:
            continue
        if st.st_size < min_bytes:
            continue
        fn = os.path.basename(p)
        text = None
        try:
            with open(p[:-4] + ".txt") as f:
                t = f.read().strip()
            if t and t != "(no speech)":
                text = t[:400]
        except OSError:
            pass
        out.append({"file": fn, "label": _capture_label(fn), "text": text,
                    "mtime": st.st_mtime, "size": st.st_size, "url": "/captures/" + fn})
    out.sort(key=lambda c: c["mtime"], reverse=True)
    # writable = the dev seat, not a presentation node (the public site is read-only for clips).
    # keyed off _IS_PRESENTATION so it matches the delete endpoint's gate (security pass 2026-07-24;
    # was `not PUSH_KEY`, which drifted loose once PUSH_KEY was removed from the Mill).
    return {"dir": CAPTURES_DIR, "captures": out[:50], "writable": not bool(SUPERVISOR_SNAPSHOT)}


def _delete_capture(fname):
    """Delete one clip from CAPTURES_DIR — basename + .mp3 only (no traversal). Driven by a dashboard
    click (build-vs-operate: the user judges a clip junk and tosses it; the server only obeys)."""
    safe = os.path.basename(fname)
    path = os.path.join(CAPTURES_DIR, safe)
    if not safe.endswith(".mp3") or not os.path.isfile(path):
        return False
    try:
        os.remove(path)
        return True
    except OSError:
        return False


def _busy_map():
    """serial -> what process is holding it RIGHT NOW (live). {} = idle. Cheap pgrep + config parse —
    sysfs can't see a libusb-held dongle, so we look at the processes. NOAA is tracked in-process;
    Airband's serials live in its -c config; a manual rtl_fm carries -d <serial>."""
    busy = {}
    if _playing() and _serial:
        busy[_serial] = "NOAA"
    try:
        out = subprocess.run(["pgrep", "-a", "rtl_airband"], capture_output=True, text=True, timeout=2).stdout
        for line in out.splitlines():
            m = re.search(r"-c\s+(\S+)", line)
            if not m:
                continue
            try:
                with open(os.path.expanduser(m.group(1))) as f:
                    for s in re.findall(r'serial\s*=\s*"([^"]+)"', f.read()):
                        busy.setdefault(s, "Airband")
            except OSError:
                pass
    except Exception:
        pass
    # rtl_fm = FM-path capture; rtl_sdr = IQ-path capture (scan-hop HOP_IQ=1, iq_channelize); rtl_tcp =
    # the retune-in-place hop (iq_hop.py opens 1001 once via rtl_tcp, retunes 452<->457 in place). All
    # carry -d <serial>. NOTE: this is process-PRESENCE only — it can't tell streaming from wedged-but-held
    # (nor would it catch a hold by an un-listed tool). True liveness = level.json freshness (_level_json);
    # wiring the supervisor's radio row to that data-flow signal is the parked supervisor rework.
    for tool in ("rtl_fm", "rtl_sdr", "rtl_tcp"):
        try:
            out = subprocess.run(["pgrep", "-a", tool], capture_output=True, text=True, timeout=2).stdout
            for line in out.splitlines():
                m = re.search(r"-d\s+(\S+)", line)
                if m:
                    busy.setdefault(m.group(1), "capture")
        except Exception:
            pass
    return busy


def _ingest_heard():
    """Append any capture not yet in HEARD_LOG (dedup by filename). Idempotent — viewing /api/heard
    keeps the durable record current, so clips can be pruned without losing the detection metadata."""
    seen = set()
    try:
        with open(HEARD_LOG) as f:
            for line in f:
                try: seen.add(json.loads(line)["file"])
                except Exception: pass
    except OSError:
        pass
    new = []
    for p in glob.glob(os.path.join(CAPTURES_DIR, "*.mp3")):
        fn = os.path.basename(p)
        if fn in seen:
            continue
        try: st = os.stat(p)
        except OSError: continue
        new.append({"ts": st.st_mtime, "ch": _capture_label(fn), "file": fn, "size": st.st_size})
    if new:
        os.makedirs(os.path.dirname(HEARD_LOG), exist_ok=True)
        with open(HEARD_LOG, "a") as f:
            for e in sorted(new, key=lambda e: e["ts"]):
                f.write(json.dumps(e) + "\n")
    return len(new)


def _heard_json(days=9, limit=8000):
    """Voice detections for the strip/correlate panels. Records are trimmed to the fields the panels
    read ({ts, ch} — strip bins by ts, correlate's voice row filters on ch): the full log rows were
    shipping ~300 KB per poll for a heatmap of counts. Time window, not row count, so the week view
    never clips; `limit` is a runaway cap; `total` stays the full count."""
    rows = _read_jsonl(HEARD_LOG)
    cutoff = time.time() - days * 86400
    window = sorted((r for r in rows if r.get("ts", 0) >= cutoff),
                    key=lambda e: e.get("ts", 0), reverse=True)[:limit]
    return {"heard": [{"ts": int(r.get("ts", 0)), "ch": r.get("ch")} for r in window], "total": len(rows)}


def _band_json(path, days=9, limit=8000):
    """Presence-burst window ({bursts, total}) for the EOT/BOT/mid strips. Trimmed to {ts} — the strip
    and correlate heatmaps bin timestamps only, and the full rows were ~400 KB per poll per band
    (the raw records live in the jsonl for anything deeper)."""
    rows = _read_jsonl(path)
    cutoff = time.time() - days * 86400
    window = sorted((r for r in rows if r.get("ts", 0) >= cutoff),
                    key=lambda e: e.get("ts", 0), reverse=True)[:limit]
    return {"bursts": [{"ts": int(r.get("ts", 0))} for r in window], "total": len(rows)}


def _eot_json(days=9, limit=8000):
    """EOT presence bursts on 457.9375 (logged by `vumon --jsonl`). Same shape as _band_json."""
    return _band_json(EOT_LOG, days, limit)


_trains_ttl = {}   # keep -> (computed_at, result): banner (keep=50) + trains panel (keep=5) both poll
                   # every 5 s; one compute (~50-150 ms) serves both for the TTL. Data changes ~20-40 s.


def _trains_json(keep=5):
    now = time.time()
    hit = _trains_ttl.get(keep)
    if hit and now - hit[0] < 4.0:
        return hit[1]
    res = _trains_json_compute(keep)
    _trains_ttl[keep] = (now, res)
    return res


def _trains_json_compute(keep=5):
    """Per-train cards from decoded EOT packets (grouped by unit address — each EOT unit = one train's
    rear device). State machine per track: LIVE/DWELLING -> DEPARTED (brakes-release signature) | PASSED
    | QUIET | LOST. Aging is HOP-AWARE (we time-share 452/457, so we miss ~half the keys — allow long
    gaps before calling a train gone). Departure leans on the release+roll signature, not on silence."""
    LIVE_S, QUIET_S = 360, 720
    # psi sanity: brake pipe tops out ~90-95; 127 = the 7-bit all-ones/invalid pattern
    # (a fuzzy BCH collision's tell — e.g. the bogus "unit 55042 psi 127 arm=Arming").
    rows = sorted((d for d in _read_jsonl(EOT_LOG)
                   if d.get("eot") and d.get("unit") is not None
                   and isinstance(d.get("pressure"), int) and 0 <= d["pressure"] <= 105
                   and d.get("ts", 0) >= time.time() - SITE_WINDOW_S),   # website shows last N days; raw log keeps ALL
                  key=lambda d: d.get("ts", 0))
    # Sessionize by unit: EOT unit numbers RECUR across trains and days, so grouping a unit's records
    # naively lumps separate passages into one card -> an inflated "dwell since hours ago" (O-7/O-8: the
    # "present since 12am" bug). A unit gone quiet longer than SESSION_GAP and back = a NEW passage = its
    # own card. This also lets distinct trains hold distinct concurrent cards (the two-track case).
    SESSION_GAP = 1800                       # 30 min quiet splits a unit into a fresh passage
    STILL_GAP   = 4 * 3600                   # ...but a STATIONARY unit decodes sparsely (weak/fuzzy
    # off-frequency bursts — O-11), so motion-0 on BOTH sides of a gap merges up to 4 h: a standing
    # train stays one passage even as its air evolves (brakes-set 59 -> cut 0 is one story, not two
    # cards). Movement on either side falls back to the tight gap, so recurring unit numbers still
    # can't lump separate passages days apart (the O-7 protection).
    by_unit = {}
    for d in rows:
        by_unit.setdefault(d["unit"], []).append(d)
    passages = []                            # [(unit, [records]), ...] — one contiguous passage each
    for unit, dl in by_unit.items():
        seg = [dl[0]]
        for d in dl[1:]:
            # one side stationary is enough: a dweller's sparse decodes gap right AT the departure
            # (76029: last standing decode 19:15, rolling decode 20:02 — same passage, one story).
            # Two moving sightings still take the tight gap, so a re-passing unit stays two passages.
            still = d.get("motion") == 0 or seg[-1].get("motion") == 0
            if d["ts"] - seg[-1]["ts"] > (STILL_GAP if still else SESSION_GAP):
                passages.append((unit, seg)); seg = [d]
            else:
                seg.append(d)
        passages.append((unit, seg))
    now = time.time()
    # cadence for the STANDING-PERSISTENCE gate (O-14): the EOT band audibly keying right now —
    # weak counts (dwellers read −45..−48). Global, not per-unit (at our range we can't split two
    # trains' keys), so a passer can briefly sustain a dweller's card — the honest limit, noted.
    _recent = [r for r in _read_jsonl(EOT_LOG) if 0 <= now - r.get("ts", 0) < 300]
    _gate = _strong_gate()                              # floor-relative "strong" gate (O-20); was absolute -42
    eot_keying = len(_recent) >= 3 or any((r.get("peak") or -99) >= _gate for r in _recent)
    cards = []
    for unit, dl in passages:
        last = dl[-1]; n = len(dl)
        ps = [d["pressure"] for d in dl]
        pmin, pmax = min(ps), max(ps)
        moving = sum(1 for d in dl if d.get("motion"))
        ever_stop = any(d.get("motion") == 0 for d in dl)
        span = last["ts"] - dl[0]["ts"]
        age = now - last["ts"]
        # --- The AIR CURVE drives the state, not the clock (O-10). A real departure = the brake pipe
        # RECHARGES back toward full AND it's rolling. An air DUMP (pressure bled to ~0 and held) is a car
        # being SET OUT — it then sits and beacons for hours (the 75976 case). NEVER read a bleed as a
        # departure; that put an "8:16 left" on a car that then sat 12 more hours. ---
        RECH = pmax - 6                              # back near the passage's own max = recharged
        DUMP = 4                                     # brake pipe at ~0 = air dumped
        dropped   = pmin <= pmax - 20                # the air came down at some point (it stopped/braked)
        recharged = dropped and last["pressure"] >= RECH and last.get("motion") == 1
        bled0     = pmin <= DUMP and last["pressure"] <= DUMP
        left_ts = None                               # the MEANINGFUL event time (roll-out or set-out), not last pkt
        cut_ts = None; ongoing = False
        if recharged:
            state, note = "departed", "recharged & rolled out"
            left_ts = next((d["ts"] for d in dl if d["pressure"] >= RECH and d.get("motion") == 1), last["ts"])
        elif bled0:
            # the air-dump is when the STANDING period BEGINS, not an end (O-11: the 76029 card read
            # "08:00→08:45 complete" while the cut train stood there all day). The card stays OPEN —
            # end = last evidence, dwell keeps counting — until the EOT has been silent 2 h (sparse
            # fuzzy decodes) or a departure signature supersedes.
            state = "setout"
            cut_ts = next((d["ts"] for d in dl if d["pressure"] <= DUMP), last["ts"])   # when the air went
            # same cadence gate as stale-dwelling: "ongoing" only while the band still keys —
            # the ONE liveness decision; the banner consumes this flag, never re-decides
            ongoing = (now - last["ts"]) < 7200 and eot_keying
            note = "air dumped %s — cut/set out, %s" % (
                time.strftime("%H:%M", time.localtime(cut_ts)),
                "STANDING (ongoing)" if ongoing else
                "stood until last heard %s" % time.strftime("%H:%M", time.localtime(last["ts"])))
            left_ts = None if ongoing else last["ts"]
        elif 0 <= age < LIVE_S:                      # live; a future-dated pkt (clock glitch, O-6) is not "live"
            if last.get("motion") == 0 and last["pressure"] >= 40:
                state, note = "dwelling", "stopped, brakes set"
            elif last.get("motion") == 0:
                state, note = "setout", "air low/bleeding — car standing (not a departure)"
            elif ever_stop:             state, note = "rolling", "moving after a stop"
            else:                       state, note = "passing", "moving through"
        elif last.get("motion") == 0 and last["pressure"] >= 40 and age < 7200 and eot_keying:
            # STANDING PERSISTENCE (the 07-06 14:00 false negative): a braked stationary train
            # decodes SPARSELY (weak off-freq bursts) — it doesn't stop existing 6 min after a
            # decode. Brakes-set + no departure signature + the band STILL KEYING = still dwelling,
            # up to 2 h. The cadence gate is the SAME one the banner uses (the owner caught them
            # disagreeing: card said dwelling on a dead band while the banner said NO) — when the
            # keys stop, this branch falls through and the card honestly reads quiet/lost.
            state = "dwelling"
            note = "standing, brakes set — sparse decodes (last %dm ago), band still keying" % (age // 60)
        elif moving > 0.8 * n and span < 600:
            state, note = "passed", "ran through, out of range"
        elif age < QUIET_S:
            state, note = "quiet", "quiet — may have missed the end"
        else:
            state, note = "lost", "gone, end unconfirmed"
        stop_ts = next((d["ts"] for d in dl if d.get("motion") == 0), last["ts"])
        end_ts  = left_ts or last["ts"]              # departed / closed setout -> the event; else last packet
        live_standing = state == "dwelling" or (state == "setout" and ongoing)
        dwell = (now - stop_ts) if live_standing else ((end_ts - stop_ts) if ever_stop else 0)
        strong = (last.get("peak") or -99) > _gate
        # crossing: only a stopped BRAKED train is BLOCKED. A set-out car we CANNOT place (no head/tail) —
        # Path A honest: flag it, don't cry blocked or clear.
        if state == "dwelling":
            crossing = "LIKELY BLOCKED — stopped %dm (%s signal)" % (int(dwell // 60), "strong" if strong else "weak")
            cflag = "blocked"
        elif state in ("rolling", "passing"):
            crossing, cflag = "occupied — train moving through", "occupied"
        elif state == "setout":
            crossing = ("standing NOW (cut/set out %dm ago) — position unknown, may or may not foul"
                        % (int((now - (cut_ts or stop_ts)) // 60))) if ongoing else \
                       "car set out (air dumped) — position unknown, may or may not foul"
            cflag = "setout"
        elif state == "departed":
            crossing, cflag = "clear — train recharged & rolled out", "clear"
        else:
            crossing, cflag = "likely clear", "clear"
        step = max(1, n // 24)
        cards.append({
            "unit": unit, "state": state, "note": note, "crossing": crossing, "cflag": cflag,
            "inferred": cflag != "clear",
            "first_ts": dl[0]["ts"], "last_ts": last["ts"], "end_ts": int(end_ts), "age": int(age), "packets": n,
            "p_first": ps[0], "p_min": pmin, "p_max": pmax, "p_last": last["pressure"],
            "motion_last": int(last.get("motion") or 0), "moving": moving,
            "marker": int(last.get("marker") or 0), "turbine": int(last.get("turbine") or 0),
            "batt": last.get("batt"), "arm": last.get("arm"), "peak": last.get("peak"),
            "dwell_sec": int(dwell), "spark": ps[::step][-24:],
            "cut_ts": (int(cut_ts) if cut_ts else None), "ongoing": ongoing,
            "all_fuzzy": all(d.get("dq") == "fuzzy" for d in dl),   # decode-quality provenance (O-11)
            "head": None,   # HOT/452 head-end: frame structure found, fields not yet decoded (RE pending)
        })
    # capture-what-we-can: HOT/head-end frames RECOVERED (undecoded) in each train's window -> card head
    hot_recs = [r for r in _read_jsonl(BOT_LOG) if r.get("hot_block")]
    for c in cards:
        # head leads the tail by 2-20 min (O-10) and a parked head polls its EOT every ~20 min while a
        # dweller's sparse fuzzy decodes leave gaps — so the join window reaches back 20 min / forward 5.
        win = [r for r in hot_recs if c["first_ts"] - 1200 <= r.get("ts", 0) <= c["last_ts"] + 300]
        if win:
            best = max(win, key=lambda r: r.get("hot_q", 0))
            addrs = sorted({r["hot_addr"] for r in win if r.get("hot_addr") is not None})
            c["head"] = {"frames": len(win), "block": best["hot_block"], "q": best.get("hot_q"),
                         "addr": best.get("hot_addr"), "addrs": addrs, "peak": best.get("peak"),
                         "last": max(r.get("ts", 0) for r in win),
                         "match": c["unit"] in addrs}   # head-end addressing THIS unit = head+tail same train

    # corroboration gate (O-11): fuzzy sync can BCH-collide on noise, minting a one-off bogus unit.
    # A card built ONLY from a single fuzzy decode needs its head vouching for it (same addr polling
    # in-window). Exact decodes and multi-record passages pass untouched.
    cards = [c for c in cards if c["packets"] >= 2 or not c.get("all_fuzzy")
             or (c["head"] and c["head"]["match"])]

    # O-11 (ground-truthed 2026-07-05): psi bled to 0 + the head STILL polling = a full train standing
    # CUT (crews cut a crossing when parking long) — not an abandoned lone car. Name the honest state:
    # the crossing may well be OPEN, but RF can't place the halves, so it stays "possibly".
    for c in cards:
        if c["state"] == "setout" and c["head"] and c["head"]["match"]:
            when = time.strftime("%H:%M", time.localtime(c["cut_ts"])) if c.get("cut_ts") else "?"
            # THE cut decision, made ONCE here (sync principle): card badge, crossing color, and the
            # banner all consume `cut`/`cflag` — nobody re-derives it from head.match downstream.
            c["cut"] = True
            c["cflag"] = "cut"
            c["note"] = ("cut %s, head still polling — train CUT & STANDING%s (not a departure)"
                         % (when, " (ongoing)" if c.get("ongoing") else ""))
            c["crossing"] = ("cut & standing%s — crossing possibly OPEN (cut), position unverified"
                             % (" NOW" if c.get("ongoing") else ", last heard %s"
                                % time.strftime("%H:%M", time.localtime(c["last_ts"]))))

    decoded = sorted(cards, key=lambda c: c["last_ts"], reverse=True)
    decoded_spans = [(c["first_ts"], c["last_ts"]) for c in decoded]
    dwell_spans = [(c["first_ts"], c["last_ts"]) for c in decoded if c["state"] == "dwelling"]
    contacts = sorted(_contacts(now, decoded_spans, dwell_spans), key=lambda c: c["last_ts"], reverse=True)
    nd = min(3, len(decoded))                            # reserve up to 3 slots for DECODED trains (the real intel)
    chosen = decoded[:nd] + contacts[:keep - nd]         # so recent unidentified contacts can't bury them
    chosen.sort(key=lambda c: c["last_ts"], reverse=True)

    # parked head-ends: an addressed unit whose HOT frames span a long time = a STATIONARY armed pair
    # nearby (a parked loco polling its EOT), not a passing train. Name it so it doesn't masquerade as one.
    by_addr = {}
    for r in hot_recs:
        a = r.get("hot_addr")
        if a is not None: by_addr.setdefault(a, []).append(r["ts"])
    parked = [{"unit": a, "frames": len(t), "span_min": int((max(t) - min(t)) // 60), "last": max(t)}
              for a, t in by_addr.items() if max(t) - min(t) >= 1800]
    parked.sort(key=lambda p: -p["last"])

    # standing signals — a band painted CONTINUOUSLY for a long stretch = a FIXED carrier, not a passing
    # train. We can't decode it (e.g. the MID/DPU off-channel carrier — characterized 2026-06-28 as a
    # continuous non-EOT-format carrier, a local source). We INFER it and card it: live until it goes quiet.
    mid = sorted((r for r in _read_jsonl(MID_LOG) if now - r.get("ts", 0) <= 86400),
                 key=lambda r: r.get("ts", 0))
    standing = []
    if mid:
        runs, cur = [], [mid[0]]
        for r in mid[1:]:
            if r["ts"] - cur[-1]["ts"] <= 300: cur.append(r)        # gaps <5min = same continuous run
            else: runs.append(cur); cur = [r]
        runs.append(cur)
        for run in runs:
            dur = run[-1]["ts"] - run[0]["ts"]
            if dur < 1800: continue                                  # >=30min continuous = a standing carrier
            pk = sorted(r["peak"] for r in run if r.get("peak") is not None)
            standing.append({
                "kind": "standing", "band": "MID · DPU watch (±12.5 kHz)",
                "onset": run[0]["ts"], "last": run[-1]["ts"], "live": now - run[-1]["ts"] <= 600,
                "dur_min": int(dur // 60), "n": len(run),
                "peak": pk[len(pk) // 2] if pk else None,
                "inference": "fixed carrier (457-side) — not a passing train; likely a railroad EOT signal booster on 457.925 MHz (FCC-reserved wayside repeater for marginal head/tail comms)",
            })
        standing.sort(key=lambda s: -s["last"])

    # per-channel readout — for EACH train card, what each of our four listening channels says about
    # THIS train, in its own window: bot (452 head), mid (±12.5k DPU body), eot (457 tail), voice. Kept
    # live (last-activity stamp per channel). The summary above stays; this is the by-band glance.
    voice_rows = _read_jsonl(HEARD_LOG)
    for c in cards:
        a, b = c["first_ts"], c["last_ts"]
        wmid = [r for r in mid if a - 60 <= r.get("ts", 0) <= b + 60]
        wvoice = [r for r in voice_rows if a - 120 <= r.get("ts", 0) <= b + 120]
        mpk = sorted(r["peak"] for r in wmid if r.get("peak") is not None)
        # mid energy in this window is booster-dominated if a standing carrier run overlaps the window
        booster = any(s["onset"] - 300 <= b and s["last"] + 300 >= a for s in standing)
        h = c.get("head")
        c["channels"] = {
            "eot": {"freq": "457.9375", "n": c["packets"], "last": c["last_ts"], "peak": c.get("peak"),
                    "moving": c["motion_last"]},
            "bot": {"freq": "452.9375", "n": (h["frames"] if h else 0),
                    "last": (h.get("last") if h else None),
                    "addr": (h.get("addr") if h else None),
                    "addrs": (h.get("addrs") if h else []),
                    "q": (h.get("q") if h else None), "peak": (h.get("peak") if h else None),
                    "match": bool(h and h.get("match"))},
            "mid": {"freq": "±12.5k", "n": len(wmid),
                    "last": (max(r.get("ts", 0) for r in wmid) if wmid else None),
                    "peak": (mpk[len(mpk) // 2] if mpk else None),
                    "pmin": (mpk[0] if mpk else None), "pmax": (mpk[-1] if mpk else None),
                    "booster": booster},
            "voice": {"freq": "airband/road", "n": len(wvoice),
                      "last": (max(r.get("ts", 0) for r in wvoice) if wvoice else None),
                      "bytes": sum(r.get("size", 0) for r in wvoice),
                      "labels": sorted({r.get("ch") for r in wvoice if r.get("ch")})[:3]},
        }
    return {"trains": chosen, "parked": parked, "standing": standing, "now": now}


STRONG_MARGIN = 11.0       # dB over the LIVE noise floor = a real, near burst. Anchors "strong" to the
                           # measured floor, not a hardcoded dB (O-20): the +10 dB antenna dropped the floor
                           # to ~-46.7, so the old absolute -42 (built for a ~-53 floor) had shrunk to ~5 dB
                           # of margin and let distant trains / booster chatter read as "near/occupied".
_STRONG_FALLBACK = -35.7   # = -46.7 + 11, the post-antenna expected gate; used only when the meter is stale
                           # (never fall back to the stale -42, which over-reports).


def _strong_gate():
    """The 'strong burst' threshold, anchored to the live noise floor (level.json). Floor ~-46.7 -> gate
    ~-35.7 now; a ~-53 floor would give ~-42 (the original value) — so this is antenna-proof by construction.
    Falls back to _STRONG_FALLBACK when the meter is stale/missing. See OBSERVATIONS O-20."""
    lv = _level_json()
    fl = lv.get("floor")
    if fl is None or lv.get("stale", True):
        return _STRONG_FALLBACK
    return round(fl + STRONG_MARGIN, 1)


def _strong_bursts(now, window_s=259200, strong=None):
    """Strong (>= `strong` dB; default = the live floor-relative gate, O-20) energy bursts on eot/bot/mid,
    tagged by band. The raw material for UNIDENTIFIED contacts (real RF, no decoded unit); files are small.

    MID/DPU is booster-aware (the O-9 fix): the 457.925 wayside booster paints a continuous carrier that
    can sit ABOVE the -42 gate for hours (measured up to ~-36 dB on 2026-07-05), gluing contact clusters
    open and vacuuming stray blips into phantom 3-band contacts. So when the MID window has enough rows
    to characterize (>=30), only excursions >= 6 dB OVER the median MID level (= the booster floor)
    count as bursts — the flat carrier itself never does. Same booster-subtraction insight as _mid_shape,
    applied one level up."""
    if strong is None: strong = _strong_gate()       # floor-relative default (O-20), not the stale -42
    out = []
    for path, band in ((EOT_LOG, "EOT"), (BOT_LOG, "BOT")):
        for d in _read_jsonl(path):
            ts = d.get("ts", 0); pk = d.get("peak")
            if pk is not None and pk >= strong and 0 <= now - ts <= window_s:
                out.append((ts, band))
    mid = [(d.get("ts", 0), d["peak"]) for d in _read_jsonl(MID_LOG)
           if d.get("peak") is not None and 0 <= now - d.get("ts", 0) <= window_s]
    if mid:
        thr = strong
        if len(mid) >= 30:
            pks = sorted(pk for _, pk in mid)
            thr = max(strong, pks[len(pks) // 2] + 6.0)   # booster floor + 6 dB
        out.extend((ts, "DPU") for ts, pk in mid if pk >= thr)
    out.sort()
    return out


def _contacts(now, decoded_spans, dwell_spans=()):
    """UNIDENTIFIED contacts: clusters of strong multi-band RF with NO decoded EOT unit — a card raised
    from energy alone ('a train was here; we couldn't ID it'). Qualifies like the banner (a strong EOT or
    BOT, or >= 2 bands; DPU-alone is not a train). Normally a window already covered by a decoded card IS
    that train, so we drop it — EXCEPT a window covered ONLY by a *dwelling* train is the OTHER track (a
    dweller holds one track; concurrent RF is a second train — the two-track case, ground-truthed
    2026-06-30). Path A (conservative): a concurrent card needs a real head/tail (EOT or BOT) present;
    MID/DPU-only during a dwell is NOT carded (it surfaces as a soft banner note), so we never fabricate a
    second train we can't actually hear. Keyed by time; upgrades to a real card once an EOT decodes."""
    CONTACT_GAP, LIVE_S = 600, 360
    dwell_set = set(dwell_spans)
    clusters = []
    for ts, band in _strong_bursts(now):
        if clusters and ts - clusters[-1]["last"] <= CONTACT_GAP:
            c = clusters[-1]; c["last"] = ts; c["bands"].add(band); c["n"] += 1
        else:
            clusters.append({"first": ts, "last": ts, "bands": {band}, "n": 1})
    cards = []
    for c in clusters:
        bands = c["bands"]
        strong_band = ("EOT" in bands) or ("BOT" in bands) or (len(bands) >= 2)
        enough = (len(bands) >= 2 and c["n"] >= 2) or c["n"] >= 4   # a real train keys many times; drop lone blips
        if not (strong_band and enough):
            continue                                     # DPU-alone, or too few bursts, is not a contact
        overlaps = [s for s in decoded_spans if not (c["last"] < s[0] or c["first"] > s[1])]
        concurrent = False
        if overlaps:
            if any(s not in dwell_set for s in overlaps):
                continue                                 # a passing/moving decoded train owns this RF
            # overlaps ONLY a dwelling train → the OTHER track. Path A: card it only on a real head/tail.
            if not (("EOT" in bands) or ("BOT" in bands)):
                continue                                 # MID/DPU-only during a dwell → soft banner note, not a card
            concurrent = True
        age = now - c["last"]; live = age < LIVE_S
        cards.append({
            "unit": None, "identified": False, "state": "unidentified", "concurrent": concurrent,
            "note": ("concurrent — 2nd head/tail during a dwell, no ID" if concurrent
                     else "detected — strong %s, no decode" % "+".join(sorted(bands))),
            "bands": sorted(bands),
            "crossing": (("2nd track — concurrent train, unidentified" if concurrent else "occupied — unidentified contact")
                         if live else
                         ("2nd track — concurrent train passed, no ID" if concurrent else "passed — unidentified, no ID")),
            "cflag": "occupied" if live else "clear", "inferred": True,
            "first_ts": c["first"], "last_ts": c["last"], "age": int(age), "packets": c["n"], "head": None,
        })
    return cards


def _tail_ts(path, nbytes=2048):
    """Last `ts` from a chronologically-appended jsonl, read cheaply from the file tail (no full scan)."""
    try:
        with open(path, "rb") as f:
            f.seek(0, 2); size = f.tell()
            f.seek(max(0, size - nbytes))
            lines = f.read().decode("utf-8", "ignore").strip().splitlines()
        for line in reversed(lines):
            try: return float(json.loads(line).get("ts", 0))
            except Exception: continue
    except OSError:
        pass
    return 0.0


def _mid_recent(now, win_s=900):
    """Recent MID (DPU ±12.5k) hits as (ts, peak) — bounded tail read."""
    out = []
    try:
        with open(MID_LOG, "rb") as f:
            f.seek(0, 2); size = f.tell(); f.seek(max(0, size - 131072))
            data = f.read().decode("utf-8", "ignore")
        for ln in data.splitlines():
            try: d = json.loads(ln)
            except Exception: continue
            ts = d.get("ts", 0); pk = d.get("peak")
            if pk is not None and 0 <= now - ts <= win_s: out.append((ts, pk))
    except OSError:
        pass
    return out


def _mid_shape(hits, K=6.0):
    """Separate the FLAT 457.925 booster floor from train-shaped EXCURSIONS above it (2026-06-30). The
    booster is a FIXED wayside repeater painting a near-constant carrier; a MOVING DPU rides ABOVE it with a
    time-envelope. The median MID peak = the booster floor (measured rock-steady at ~-40 dB over a 2 h dwell).
    States: quiet | booster_only (flat carrier, nothing above) | excursion (strong but brief/scattered —
    weak/edge, unconfirmed) | train_shaped (a SUSTAINED excursion above the floor = candidate MOVING source).
    Honest by construction (Path A): 'candidate', never a claimed train — MID carries no decodable ID."""
    recent = [(ts, pk) for ts, pk in hits if pk is not None]
    if len(recent) < 6:
        return {"state": "quiet", "floor": None, "n": len(recent), "over": 0.0, "sustained": False}
    pks = sorted(pk for _, pk in recent)
    floor = pks[len(pks) // 2]                       # median = the booster carrier level
    exc = [ts for ts, pk in recent if pk > floor + K]
    over = max(pk - floor for _, pk in recent)
    sustained = len(exc) >= 4 and (max(exc) - min(exc)) >= 120   # excursions spread over >=2 min = a run, not a blip
    if over < K or not exc:
        state = "booster_only"
    elif sustained:
        state = "train_shaped"
    else:
        state = "excursion"
    return {"state": state, "floor": round(floor, 1), "n": len(recent),
            "n_exc": len(exc), "over": round(over, 1), "sustained": sustained}


def _istrain_json():
    """THE headline inference: is a train near the crossing RIGHT NOW? Layers, most-certain first:
    a decoded train card (YES) -> strong undecoded telemetry RF (LIKELY) -> deaf capture (NOT SURE) -> NO.
    The EOT/BOT/MID telemetry is the KEY signal, and a burst must be STRONG (>= floor + 11 dB) to count — so
    the weak MID noise blips (the O-1 false positives) can't trip it. Voice is sporadic background: it NEVER
    triggers, only corroborates (a voice spike upgrades the detail). Inferred."""
    now = time.time()
    ACT_S = 300                      # "recent" window, seconds (hop-aware-ish)
    STRONG = _strong_gate()          # floor-relative "real burst" gate (O-20); was absolute -42 (stale post-antenna)
    VOICE_SPIKE = 4                  # this many voice clips in the window = "a lot" (vs the sporadic baseline)
    tj = _trains_json(keep=50)                          # look past the 3-card display trim to find live / last-decoded
    cards, parked = tj["trains"], tj["parked"]
    last_train = next((c["last_ts"] for c in cards if c.get("unit") is not None), 0)   # last DECODED train, not a contact
    streaming = not _level_json().get("stale", True)

    def chan(path, win_bytes=65536):
        """(strongest peak in window | None, count in window) — bounded tail read (logs append in order)."""
        best = None; n = 0
        try:
            with open(path, "rb") as f:
                f.seek(0, 2); size = f.tell(); f.seek(max(0, size - win_bytes))
                data = f.read().decode("utf-8", "ignore")
            for ln in data.splitlines():
                try: d = json.loads(ln)
                except Exception: continue
                if 0 <= now - d.get("ts", 0) < ACT_S:
                    n += 1
                    pk = d.get("peak")
                    if pk is not None and (best is None or pk > best): best = pk
        except OSError:
            pass
        return best, n

    eot_pk, eot_n = chan(EOT_LOG); bot_pk, _ = chan(BOT_LOG); mid_pk, _ = chan(MID_LOG); _, voice_n = chan(HEARD_LOG)
    # THE SYNC PRINCIPLE (the owner, 2026-07-06: "the cards that are still live need to be in sync with
    # the banner — that's the whole point"): every liveness decision is made ONCE, in the card state
    # machine (incl. the cadence-gated standing persistence). The banner consumes card states verbatim
    # and re-decides NOTHING — so card and banner cannot disagree by construction.
    live = [c for c in cards if c["state"] in ("dwelling", "rolling", "passing")]
    s_eot = eot_pk is not None and eot_pk >= STRONG
    s_bot = bot_pk is not None and bot_pk >= STRONG
    s_mid = mid_pk is not None and mid_pk >= STRONG
    hot = [n for n, s in (("EOT", s_eot), ("BOT", s_bot), ("DPU", s_mid)) if s]
    mshape = _mid_shape(_mid_recent(now))            # is the MID a flat booster, or a train-shaped excursion above it?
    # LIKELY needs a definitive head/tail (EOT or BOT = the train's own device keying in range), OR >=2 bands
    # together (real co-occurrence). DPU/MID ALONE does NOT qualify: it's the trip-happy channel and carries
    # from trains that aren't here — the telemetry only means a train when the channels agree.
    likely = s_eot or s_bot or len(hot) >= 2

    def ago(s):
        s = int(max(0, s))
        return ("%dm" % (s // 60)) if s < 3600 else ("%dh %dm" % (s // 3600, (s % 3600) // 60))

    if live:
        dwell = [c for c in live if c["state"] == "dwelling"]
        if dwell:
            c = dwell[0]
            # concurrent traffic on the OTHER track: a 2nd train carded (head/tail), or — softer — sustained
            # MID/DPU now (booster-or-2nd-train, can't tell without a head/tail). Path A: name it, don't overclaim.
            if any(cc.get("concurrent") for cc in cards):
                extra = " · + concurrent train, other track"
            elif mshape["state"] == "train_shaped":
                extra = " · candidate 2nd train — train-shaped MID excursion (+%gdB over booster, unverified)" % mshape["over"]
            elif mshape["state"] == "excursion":
                extra = " · MID excursion over booster (weak/brief — unconfirmed)"
            elif s_mid:
                extra = " · MID = booster carrier (flat ~%gdB)" % (mshape.get("floor") or 0)
            else:
                extra = ""
            # the banner REPRESENTS the card's confidence (the owner 2026-07-06): a persisted dwell
            # (sparse decodes, band keying) must not read as certain as a fresh one — say the grade
            fresh = "" if c["age"] < 360 else " · last decode %dm ago, band still keying" % (c["age"] // 60)
            return {"verdict": "YES", "level": "blocking", "headline": "TRAIN — crossing likely BLOCKED",
                    "detail": "unit %s stopped %dm%s%s" % (c["unit"], c["dwell_sec"] // 60, fresh, extra), "inferred": True, "now": now}
        c = live[0]
        return {"verdict": "YES", "level": "train", "headline": "TRAIN PASSING",
                "detail": "unit %s moving through" % c["unit"], "inferred": True, "now": now}
    # setout liveness = the card's own cadence-gated "ongoing" flag (decided in the state machine,
    # nowhere else — the sync principle). Head-still-polling upgrades it to the O-11 ground-truthed
    # read: a full train standing CUT (crossing possibly open).
    setout = [c for c in cards if c["state"] == "setout" and c.get("ongoing")]
    if setout:
        c = setout[0]
        standing = (" — standing %s" % ago(c["dwell_sec"])) if c.get("dwell_sec") else ""
        fresh = "" if c["age"] < 360 else " · last decode %dm ago, band still keying" % (c["age"] // 60)
        if c.get("cut"):
            # CUT gets its OWN level, not setout's clothes (the owner 2026-07-08, after the second eyes-on
            # confirm — O-11 76029 + O-21 65793, crossing open both times). Distinct banner color/tag;
            # wording stays honest: RF can't place the gap, so "possibly OPEN" until eyes say otherwise.
            # `cut` comes from the card state machine — the ONE decision (sync principle), not re-derived.
            return {"verdict": "YES", "level": "cut", "headline": "TRAIN STANDING — CUT (air dumped)",
                    "detail": "unit %s air dumped, head still polling%s%s — crossing possibly OPEN (cut), unverified"
                              % (c["unit"], standing, fresh),
                    "inferred": True, "now": now}
        return {"verdict": "MAYBE", "level": "setout", "headline": "CAR SET OUT — crossing status unknown",
                "detail": "unit %s air dumped & standing%s%s — no head/tail to say if it fouls" % (c["unit"], standing, fresh),
                "inferred": True, "now": now}
    # DWELLING JOIN (O-11 fallback, when nothing decodes at all): a parked head actively polling its EOT
    # + a live strong EOT-answer cadence = a stationary train is there, even with zero decoded packets.
    # (The 07-04 74811 and 07-05 76029 dwells both looked exactly like this for hours.) Honest scope:
    # the train is real; its unit is the HEAD's addr; crossing state is unknown without psi.
    poll = next((p for p in parked if now - p["last"] <= 1200), None)
    if poll and s_eot and eot_n >= 3:
        return {"verdict": "YES", "level": "dwelling", "headline": "TRAIN DWELLING — head parked & polling",
                "detail": "head unit %s polling ~%dm + strong EOT cadence — tail undecoded; crossing state unknown"
                          % (poll["unit"], poll["span_min"]),
                "inferred": True, "now": now}
    if likely:                                    # head/tail or co-occurring telemetry — voice only corroborates
        bits = "+".join(hot)
        if len(hot) >= 2: bits += " together"
        if voice_n >= VOICE_SPIKE: bits += ", heavy voice"
        return {"verdict": "LIKELY", "level": "likely", "headline": "LIKELY — train-signature RF",
                "detail": "strong %s, not decoded yet" % bits, "inferred": True, "now": now}
    if not streaming:
        return {"verdict": "NOT SURE", "level": "unsure", "headline": "NOT SURE — listening is quiet/down",
                "detail": "capture not streaming — can't tell", "inferred": True, "now": now}
    # NO — but don't silently swallow strong-but-lone DPU; name it by SHAPE, not a train.
    if mshape["state"] == "train_shaped":
        detail = "MID excursion above the booster — candidate moving source, no head/tail (train-shaped, unverified)"
    elif s_mid:
        detail = "DPU/mid = booster carrier (flat ~%gdB) — no head/tail" % (mshape.get("floor") or 0)
    else:
        detail = ("clear — last train %s ago" % ago(now - last_train)) if last_train else "clear — none seen yet"
    return {"verdict": "NO", "level": "clear", "headline": "NO TRAIN", "detail": detail, "inferred": True, "now": now}


def _version_json():
    """Commit id for the header. VM: read it from the git checkout serving this ('*' appended when the
    working tree has uncommitted changes — so local iteration is visibly not the committed build). NAS:
    the app dir is curl-fetched files, not a checkout — fetch-app.sh stamps dashboard/VERSION from the
    a git host API and we read that. The VERSION file wins if present (a checkout won't have one)."""
    try:
        with open(os.path.join(HERE, "VERSION")) as f:
            v = f.read().strip()
        if v:
            return {"commit": v}
    except OSError:
        pass
    try:
        repo = os.path.dirname(HERE)
        sha = subprocess.run(["git", "-C", repo, "rev-parse", "--short", "HEAD"],
                             capture_output=True, text=True, timeout=3).stdout.strip()
        if sha:
            dirty = subprocess.run(["git", "-C", repo, "status", "--porcelain"],
                                   capture_output=True, text=True, timeout=3).stdout.strip()
            return {"commit": sha + ("*" if dirty else "")}
    except Exception:
        pass
    return {"commit": None}


def _level_json():
    """vumon's live instantaneous level (the meter polls this fast). Stale if no fresh write in 3s
    (hopping between bands, idle, or capture stopped)."""
    try:
        with open(LEVEL_LOG) as f:
            d = json.load(f)
    except Exception:
        return {"stale": True}
    age = time.time() - d.get("ts", 0)
    d["age"] = round(age, 1)
    d["stale"] = age > 3.0
    return d


def _correlate_json(window_min=30):
    """Multi-band train-activity correlation (read-only). Runs scripts/correlate.py's engine — no radio touch."""
    try:
        import sys as _sys
        sp = os.path.join(HERE, "..", "scripts")
        if sp not in _sys.path:
            _sys.path.insert(0, sp)
        import correlate as _c
        return _c.correlate(window_min=window_min)
    except Exception as e:
        return {"now": time.time(), "window_min": window_min, "state": "error",
                "bands": [], "events": [], "reasons": ["correlate engine error: %s" % e]}


# ── Supervisor registry ──────────────────────────────────────────────────────────────────────────
# SERVICES we manage (modular: add a tuple → it shows up + gets controls). Dongles are added live from
# the radios scan as "resource" units. (job = one-line "what it does".)
SUP_SERVICES = [
    # the HONEST list (reviewed): every row is a real unit doing real work.
    # Retired the same day: istrain-eot-reap.* (dead janitor — pruning is DATA.md's deliberate future;
    # reaper.py the FILE stays, eot-decode imports its decode adapter) and istrain-eot-watch (the old
    # single-freq 457 dwell, superseded by scan-hop — a dongle-1001 conflict landmine if ever started).
    ("istrain-dash",        "Dashboard",       "serves this dashboard (VM :8157 local preview)"),
    ("istrain-scan-hop",    "UHF · 1001",      "dongle 1001: 452 ⇄ 457 retune-in-place hop → BOT/EOT/MID logs + clips + live meter"),
    ("istrain-airband",     "Voice · 1002",    "dongle 1002: rail voice, 5 NS channels → mp3 clips + heard log"),
    ("istrain-eot-decode",  "EOT · decode",    "CPU-only (no radio): EOT clips → unit/psi/motion, two-pass → enriches eot.jsonl; never culls"),
    ("istrain-bot-recover", "BOT · decode",    "CPU-only (no radio): BOT clips → HOT frame + head address → enriches bot.jsonl; never culls"),
    ("istrain-push",        "NAS · push",      "pushes logs + supervisor snapshot to the NAS every ~20s — the public site's data lifeline"),
    ("istrain-trim.timer",  "Captures · trim", "hourly: prunes /tmp/airband mp3 audio >24h (the ONLY deleter; heard.jsonl kept)"),
]
_SUP_CTL_OK = {u[0] for u in SUP_SERVICES}   # only registered services may be controlled


_cpu_seen = {}   # process name -> (wallclock, cputime_s) — the heartbeat's previous sample


def _cpu_heartbeat(procname, now):
    """CPU-seconds-per-wall-second for `procname` since the last poll — the passive liveness proxy
    for processes with no data artifact (airband). None = no baseline yet (first poll / new pid)."""
    try:
        pid = int(subprocess.run(["pgrep", "-x", "-o", procname], capture_output=True,
                                 text=True, timeout=2).stdout.split()[0])
        with open("/proc/%d/stat" % pid) as f:
            parts = f.read().split()
        cpu = (int(parts[13]) + int(parts[14])) / os.sysconf("SC_CLK_TCK")
    except Exception:
        return None
    prev = _cpu_seen.get(procname)
    _cpu_seen[procname] = (now, cpu)
    if not prev or now - prev[0] < 5 or cpu < prev[1]:   # too soon, or a new pid reset the counter
        return None
    return (cpu - prev[1]) / (now - prev[0])


def _svc_show(unit):
    try:
        out = subprocess.run(["systemctl", "--user", "show", unit,
                              "--property=ActiveState,SubState,LoadState,ActiveEnterTimestamp"],
                             capture_output=True, text=True, timeout=3).stdout
        return dict(l.split("=", 1) for l in out.splitlines() if "=" in l)
    except Exception:
        return {}


def _supervisor_json():
    busy = _busy_map()
    try:
        airband_alive = subprocess.run(["pgrep", "-x", "rtl_airband"], capture_output=True).returncode == 0
    except Exception:
        airband_alive = False
    units = []
    for unit, name, job in SUP_SERVICES:
        kv = _svc_show(unit)
        active = kv.get("ActiveState", "unknown")
        dot = "listening" if active == "active" else "offline" if active == "failed" else "ready"
        detail = kv.get("SubState") or active
        if unit == "istrain-airband" and active != "active" and airband_alive:
            dot, detail = "listening", "running (manual)"     # run by hand for now; service still off
        units.append({"id": unit, "name": name, "kind": "service", "job": job, "dot": dot,
                      "state": active, "detail": detail, "since": kv.get("ActiveEnterTimestamp", ""),
                      "can_ctl": kv.get("LoadState") == "loaded"})
    try:
        with open(os.path.join(HERE, "radios.json")) as f:
            radios = json.load(f).get("radios", [])
    except Exception:
        radios = []
    now = time.time()
    for r in radios:
        s = r.get("serial"); role = (r.get("role") or "").lower()
        present, dvb = _dongle_state(s)   # live sysfs read — radios.json's flags go stale on unplug
        if not present:                dot, detail = "offline", "no radio — not detected"
        elif dvb:                      dot, detail = "blocked", "DVB-claimed"
        elif s in busy:
            # truth by DATA FLOW (the 06-24/06-25 lesson): USB-enumerated + process-held is NOT alive —
            # a wedged dongle stays enumerated and held while zero samples flow. Green only on evidence.
            holder = busy[s]
            if "voice" in role or "160" in role:
                # voice liveness, two proofs (truth-by-data-flow): a recent clip is the strongest;
                # else the CPU HEARTBEAT (the CPU-dead question): a live
                # 5-channel demod burns steady CPU (~4-5%), a stalled USB stream burns ~none. So a
                # quiet band shows green "demodulating" instead of gray UNKNOWN, and a true stall
                # shows orange STALLED — the missing 06-25 heartbeat, measured not guessed.
                mts = [os.path.getmtime(p) for p in glob.glob(os.path.join(CAPTURES_DIR, "*.mp3"))[:400]]
                last = max(mts, default=0)
                ago = ("last clip %dm ago" % ((now - last) // 60)) if last else "no clip this session"
                if last and now - last < 1800:
                    dot, detail = "listening", "%s · capturing (%s)" % (holder, ago)
                else:
                    hb = _cpu_heartbeat("rtl_airband", now)
                    if hb is None:
                        dot, detail = "held", ("%s holds it — %s; measuring CPU heartbeat…" % (holder, ago))
                    elif hb >= 0.01:
                        dot, detail = "listening", ("%s · demodulating (%.0f%% CPU) — band quiet, %s"
                                                    % (holder, hb * 100, ago))
                    else:
                        dot, detail = "stalled", ("%s holds it but ~0%% CPU — USB stream stalled? replug/reset (%s)"
                                                  % (holder, ago))
            else:
                # the UHF hop writes level.json ~5 Hz while samples flow — freshness IS the liveness
                lv = _level_json(); age = lv.get("age")
                if age is not None and age < 30:
                    dot, detail = "listening", "%s · streaming (level %.0fs fresh)" % (holder, age)
                else:
                    dot, detail = "stalled", ("%s holds it but NO samples flow (level.json %s) — wedged? replug/reset"
                                              % (holder, ("%.0fs stale" % age) if age is not None else "absent"))
        else:                          dot, detail = "ready", "idle"
        units.append({"id": s, "name": f"{s} · {r.get('role','')}", "kind": "resource", "job": r.get("role", ""),
                      "dot": dot, "state": dot, "detail": detail, "since": "", "can_ctl": False})
    import time as _t
    return {"units": units, "ts": _t.time(), "controls": True}


# The NAS is the PRESENTATION node (SUPERVISOR_SNAPSHOT set): no systemd, no dongles, no pkill — its
# control endpoints can't execute anything themselves. Control model (the owner, 2026-07-05):
#   · the panel HIDES all control buttons behind a triple-tap unlock on BOTH sites (identical display;
#     no password — obscurity is the accepted bar: "a normal person should not be able to stop a
#     service without knowing where to look").
#   · with CTL_FORWARD set (the VM dash's tailnet URL), the NAS FORWARDS ctl/reset POSTs to the VM —
#     so the unlocked live site really operates the rig from anywhere.
#   · ctl/reset POSTs require the X-Ops header (the unlocked panel sends it) — a speed bump so blind
#     curl scanners bounce with 403. NOAA never forwards: its audio plays on the VM's speaker.
_IS_PRESENTATION = bool(SUPERVISOR_SNAPSHOT)
CTL_FORWARD = os.environ.get("CTL_FORWARD", "")     # e.g. http://dev-seat:8157 (NAS compose only)


def _supervisor_snapshot_or_live():
    """NAS (presentation node): serve the VM-pushed snapshot if configured + present. VM: run live.
    The NAS has no systemd/dongles, so a live read there is meaningless — it shows the rig's pushed state.
    controls = this node can execute (VM) or forward (NAS w/ CTL_FORWARD); noaa = radios are local."""
    if SUPERVISOR_SNAPSHOT and os.path.exists(SUPERVISOR_SNAPSHOT):
        try:
            with open(SUPERVISOR_SNAPSHOT) as f:
                d = json.load(f)
            d["controls"] = bool(CTL_FORWARD)
            d["noaa"] = False
            return d
        except Exception:
            pass
    d = _supervisor_json()
    if _IS_PRESENTATION:
        d["controls"] = bool(CTL_FORWARD)
        d["noaa"] = False
    else:
        d["noaa"] = True
    return d


def _ctl_forward(path_qs):
    """Relay a control POST to the VM dash over the tailnet. Returns (json-dict, http-code)."""
    import urllib.request
    req = urllib.request.Request(CTL_FORWARD.rstrip("/") + path_qs, method="POST",
                                 headers={"X-Ops": "1"}, data=b"")
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return json.loads(r.read().decode() or "{}"), r.status
    except Exception as e:
        return {"ok": False, "msg": "forward to VM failed: %s" % e}, 502


def _log_error(source, msg, kind="error"):
    """Append one line to the errors log (the Errors card's substrate). SELF-PRUNING — unlike the
    detection logs (keep-all, deliberate), this is operational noise with a public feeder (client
    reports): past ~256 KB it rewrites itself keeping the newest 1000 lines. Never raises."""
    try:
        os.makedirs(os.path.dirname(ERRORS_LOG), exist_ok=True)
        with open(ERRORS_LOG, "a") as f:
            f.write(json.dumps({"ts": time.time(), "source": source, "kind": kind,
                                "msg": str(msg)[:500]}) + "\n")
        if os.path.getsize(ERRORS_LOG) > 256 * 1024:
            with open(ERRORS_LOG) as f:
                lines = f.readlines()[-1000:]
            tmp = ERRORS_LOG + ".tmp"
            with open(tmp, "w") as f:
                f.writelines(lines)
            os.replace(tmp, ERRORS_LOG)
    except OSError:
        pass


def _errors_json(limit=100):
    """The Errors card: LIVE issues right now (failed units, non-green dongles) + recent history
    (this node's errors.jsonl merged with the VM's pushed copy on the NAS)."""
    rows = list(_read_jsonl(ERRORS_LOG))
    if ERRORS_PUSHED:
        rows = rows + _read_jsonl(ERRORS_PUSHED)
    rows.sort(key=lambda r: r.get("ts", 0), reverse=True)
    live = []
    try:
        for u in _supervisor_snapshot_or_live().get("units", []):
            # 'held' (voice liveness-unknown) is not an issue — quiet band ≠ dead (the 06-25 lesson)
            if u.get("dot") in ("offline", "blocked", "stalled") or u.get("state") == "failed":
                live.append({"id": u.get("id"), "name": u.get("name"),
                             "dot": u.get("dot"), "detail": u.get("detail")})
    except Exception:
        pass
    return {"now": time.time(), "live": live, "history": rows[:limit], "total": len(rows)}


_err_report_ts = []          # client-report rate limit (public endpoint): 20/min then drop


def _client_error_report(body):
    global _err_report_ts
    now = time.time()
    _err_report_ts = [t for t in _err_report_ts if now - t < 60]
    if len(_err_report_ts) >= 20:
        return False
    _err_report_ts.append(now)
    try:
        d = json.loads(body.decode("utf-8", "ignore") or "{}")
    except Exception:
        return False
    msg = str(d.get("msg", ""))[:500]
    src = str(d.get("src", ""))[:200]
    line = d.get("line", 0)
    if not msg:
        return False
    _log_error("client", "%s (%s:%s)" % (msg, src, line))
    return True


def _supervisor_ctl(unit, action):
    """start/stop/restart a REGISTERED service only (whitelist) — the friendly face on systemctl."""
    if unit not in _SUP_CTL_OK:
        return False, "unit not controllable"
    if action not in ("restart", "start", "stop"):
        return False, "bad action"
    try:
        # --no-block (2026-07-06, learned from the live hard-reset test): a plain restart BLOCKS until
        # the unit is fully up — scan-hop's wedged stop ran 90 s, our 12 s client timeout killed the
        # systemctl mid-job and STRANDED the decoders (stopped, never started). Enqueue-and-return:
        # systemd owns the job to completion; the supervisor dots show the truth on the next poll.
        r = subprocess.run(["systemctl", "--user", "--no-block", action, unit],
                           capture_output=True, text=True, timeout=12)
        ok, msg = (r.returncode == 0), (r.stderr.strip() or f"{action} requested — watch the row's dot")
    except Exception as e:
        ok, msg = False, str(e)
    # audit trail for the Errors card: every control action is an event; a failure is an error
    _log_error("supervisor", "%s %s: %s" % (action, unit, msg), kind=("event" if ok else "error"))
    return ok, msg


def _supervisor_reset():
    """The hard reset: free any stuck manual captures (pkill), then restart every managed helper EXCEPT the
    dashboard (it's serving this request). Knows the services + the capture scripts; device-level DVB/USB
    resets need sudo and stay out of scope — surfaced via the dongle rows, run from cmd-palette."""
    _log_error("supervisor", "HARD RESET invoked — pkill captures + restart all helpers", kind="event")
    killed = []
    for p in ("rtl_airband", "rtl_fm", "rtl_sdr", "rtl_tcp"):   # rtl_tcp/rtl_sdr = the weighted hop's tooling
        if subprocess.run(["pkill", "-x", p], capture_output=True).returncode == 0:
            killed.append(p)
    time.sleep(2)                                   # let the dongles release
    out = []
    for unit, name, job in SUP_SERVICES:
        if unit == "istrain-dash":
            continue
        ok, msg = _supervisor_ctl(unit, "restart")
        out.append({"unit": unit, "ok": ok, "msg": msg})
    return {"killed": killed, "reset": out}


def _hits_load():
    try:
        with open(HITS_LOG) as f:
            return json.load(f)
    except Exception:
        return {"total": 0, "ips": []}


def _hits_json():
    d = _hits_load()
    return {"unique": len(set(d.get("ips", []))), "total": d.get("total", 0)}


def _record_hit(ip):
    """One page load: bump total, add the (hashed) visitor to the unique set, persist atomically."""
    h = hashlib.sha1(ip.encode()).hexdigest()[:16]
    with _hits_lock:
        d = _hits_load()
        d["total"] = d.get("total", 0) + 1
        ips = set(d.get("ips", [])); ips.add(h)
        d["ips"] = sorted(ips)
        try:
            os.makedirs(os.path.dirname(HITS_LOG), exist_ok=True)
            tmp = HITS_LOG + ".tmp"
            with open(tmp, "w") as f:
                json.dump(d, f)
            os.replace(tmp, HITS_LOG)
        except OSError:
            pass


class Handler(SimpleHTTPRequestHandler):
    def __init__(self, *a, **kw):
        super().__init__(*a, directory=HERE, **kw)

    def end_headers(self):
        # The dashboard iterates fast and sits behind a Cloudflare tunnel that edge-caches .js/.css by
        # extension — which serves a STALE app after a deploy (cost us the panels + the hit beacon). Tell
        # CF/browsers never to store the app shell + API so a deploy shows up immediately. (mp3 captures
        # under /captures/ are left cacheable — they're immutable and big.)
        p = urlparse(self.path).path
        if p == "/" or p.endswith((".js", ".css", ".html")) or p.startswith("/api/"):
            self.send_header("Cache-Control", "no-store")
        super().end_headers()

    def _win_days(self):
        """Optional ?hours=N window on the band endpoints (the 1-min zoom strips pass hours=6 so the
        server only ships what gets drawn); default stays the 9-day week view (correlate's heatmap)."""
        q = parse_qs(urlparse(self.path).query)
        try:
            h = float(q.get("hours", [0])[0])
        except (ValueError, TypeError):
            h = 0
        return (max(0.25, min(240, h)) / 24.0) if h else 9

    def _client_ip(self):
        # behind Cloudflare the socket IP is Cloudflare's — the real visitor is in CF-Connecting-IP
        return (self.headers.get("CF-Connecting-IP")
                or self.headers.get("X-Forwarded-For", "").split(",")[0].strip()
                or self.client_address[0])

    def _send_json(self, obj, code=200):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _send_capture(self, fname):
        """Serve one MP3 clip from CAPTURES_DIR — basename + .mp3 only (no traversal). Supports HTTP
        Range so <audio> can load metadata/seek — without it WebKit spins forever 'loading'."""
        safe = os.path.basename(fname)
        path = os.path.join(CAPTURES_DIR, safe)
        if not safe.endswith(".mp3") or not os.path.isfile(path):
            return self.send_error(404)
        try:
            size = os.path.getsize(path)
            start, end, partial = 0, size - 1, False
            rng = self.headers.get("Range", "")
            if rng.startswith("bytes="):
                partial = True
                s, _, e = rng[6:].partition("-")
                start = int(s) if s else 0
                end = min(int(e) if e else size - 1, size - 1)
                if start > end or start >= size:
                    self.send_response(416)
                    self.send_header("Content-Range", f"bytes */{size}")
                    self.end_headers()
                    return
            with open(path, "rb") as f:
                f.seek(start)
                data = f.read(end - start + 1)
        except OSError:
            return self.send_error(404)
        self.send_response(206 if partial else 200)
        self.send_header("Content-Type", "audio/mpeg")
        self.send_header("Accept-Ranges", "bytes")
        if partial:
            self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        path = urlparse(self.path).path
        if path == "/api/hits":
            return self._send_json(_hits_json())
        if path == "/api/hit":
            # client beacon: dedup unique visitors by a persistent localStorage token, NOT by IP —
            # behind the Cloudflare tunnel the real per-visitor IP doesn't reach us (all collapse to one).
            vid = (parse_qs(urlparse(self.path).query).get("vid") or [""])[0]
            if 8 <= len(vid) <= 80 and re.fullmatch(r"[0-9a-fA-F-]+", vid):
                _record_hit(vid)
            return self._send_json(_hits_json())
        if path == "/api/noaa":
            return self._send_json(_state_json())
        if path == "/api/captures":
            return self._send_json(_captures_json())
        if path == "/api/activity":
            return self._send_json({"busy": _busy_map()})
        if path == "/api/heard":
            return self._send_json(_heard_json())  # durable voice-activity log ONLY (for the correlate signal); ingest runs on a timer, not per-request — no clip-dir glob on every poll
        if path == "/api/eot":
            return self._send_json(_eot_json(days=self._win_days()))
        if path == "/api/trains":
            return self._send_json(_trains_json())
        if path == "/api/istrain":
            return self._send_json(_istrain_json())
        if path == "/api/correlate":
            q = parse_qs(urlparse(self.path).query)
            try: win = int(q.get("window", ["30"])[0])
            except Exception: win = 30
            win = max(5, min(720, win))
            return self._send_json(_correlate_json(win))
        if path == "/api/bot":
            return self._send_json(_band_json(BOT_LOG, days=self._win_days()))
        if path == "/api/midtrain":
            return self._send_json(_band_json(MID_LOG, days=self._win_days()))
        if path == "/api/level":
            return self._send_json(_level_json())
        if path == "/api/version":
            return self._send_json(_version_json())
        if path == "/api/errors":
            return self._send_json(_errors_json())
        if path == "/api/supervisor":
            return self._send_json(_supervisor_snapshot_or_live())
        if path.startswith("/captures/"):
            return self._send_capture(path[len("/captures/"):])
        return super().do_GET()

    def do_POST(self):
        path = urlparse(self.path).path
        if path in ("/api/supervisor/ctl", "/api/supervisor/reset"):
            # X-Ops = the unlocked panel's marker; blind curl pokes bounce (speed bump, not a password)
            if self.headers.get("X-Ops") != "1":
                return self._send_json({"ok": False, "msg": "controls are gated — unlock the supervisor panel"}, code=403)
            if _IS_PRESENTATION:
                if CTL_FORWARD:
                    body, code = _ctl_forward(self.path)
                    return self._send_json(body, code=code)
                return self._send_json({"ok": False, "msg": "presentation node — no control path to the VM configured"}, code=403)
        if _IS_PRESENTATION and path == "/api/noaa":
            # NOAA audio plays on the VM's own speaker — meaningless remotely; never forwarded
            return self._send_json({"ok": False, "msg": "NOAA test is VM-local"}, code=403)
        if path == "/api/noaa":
            q = parse_qs(urlparse(self.path).query)
            serial = (q.get("serial") or [NOAA_DEFAULT])[0]
            err = _toggle(serial)
            return self._send_json(_state_json(error=err), code=200 if not err else 409)
        if path == "/api/captures/delete":
            # read-only on a presentation node (the public site): clip deletion is a build-vs-operate
            # action for the dev seat only. Matches the panel's writable=false. (security pass 2026-07-24)
            if _IS_PRESENTATION:
                return self._send_json({"ok": False, "msg": "read-only node"}, code=403)
            q = parse_qs(urlparse(self.path).query)
            ok = _delete_capture((q.get("file") or [""])[0])
            return self._send_json({"deleted": ok}, code=200 if ok else 404)
        if path == "/api/supervisor/ctl":
            q = parse_qs(urlparse(self.path).query)
            ok, msg = _supervisor_ctl((q.get("unit") or [""])[0], (q.get("action") or [""])[0])
            return self._send_json({"ok": ok, "msg": msg}, code=200 if ok else 400)
        if path == "/api/supervisor/reset":
            return self._send_json(_supervisor_reset())
        if path == "/api/push":
            q = parse_qs(urlparse(self.path).query)
            ok, msg, code = self._push_receive((q.get("log") or [""])[0], (q.get("key") or [""])[0])
            return self._send_json({"ok": ok, "msg": msg}, code=code)
        if path == "/api/push-clip":
            # voiced clip push (2026-07-10): the VM ships real-speech mp3s (+ .txt sidecars) so the
            # public captures panel has audio to play. Same PUSH_KEY gate as /api/push.
            q = parse_qs(urlparse(self.path).query)
            ok, msg, code = self._push_clip_receive((q.get("file") or [""])[0], (q.get("key") or [""])[0])
            return self._send_json({"ok": ok, "msg": msg}, code=code)
        if path == "/api/errors/report":
            # browser JS errors → the Errors card. Public by design (a broken page on the owner's phone
            # is exactly the shareable error); rate-limited + length-capped server-side.
            try:
                n = min(int(self.headers.get("Content-Length", 0)), 4096)
            except ValueError:
                n = 0
            ok = _client_error_report(self.rfile.read(n) if n > 0 else b"")
            return self._send_json({"ok": ok})
        self.send_error(404)

    def _push_clip_receive(self, fname, key):
        """Receive one voiced clip (mp3 or its .txt sidecar) into CAPTURES_DIR. PUSH_KEY-gated,
        basename-only (no traversal), size-capped, atomic write. Prunes CAPTURES_DIR to the newest
        ~120 files so the presentation node's clip shelf is self-limiting. (2026-07-10)"""
        if not PUSH_KEY:    return False, "push disabled (no PUSH_KEY on this node)", 403
        if key != PUSH_KEY: return False, "bad key", 403
        safe = os.path.basename(fname)
        if not (safe.endswith(".mp3") or safe.endswith(".txt")) or safe.startswith("."):
            return False, "mp3/txt only", 400
        try:
            n = int(self.headers.get("Content-Length", 0))
        except ValueError:
            n = 0
        if n <= 0 or n > 8 * 1024 * 1024:
            return False, "empty or oversized body", 400
        body = self.rfile.read(n)
        try:
            os.makedirs(CAPTURES_DIR, exist_ok=True)
            tmp = os.path.join(CAPTURES_DIR, safe + ".tmp")
            with open(tmp, "wb") as f:
                f.write(body)
            os.replace(tmp, os.path.join(CAPTURES_DIR, safe))
            files = sorted(glob.glob(os.path.join(CAPTURES_DIR, "*.mp3")) +
                           glob.glob(os.path.join(CAPTURES_DIR, "*.txt")),
                           key=os.path.getmtime, reverse=True)
            for old in files[120:]:
                try: os.remove(old)
                except OSError: pass
        except OSError as e:
            return False, f"write failed: {e}", 500
        return True, "ok", 200

    def _push_receive(self, name, key):
        """Receive a full-file replace from the VM pusher (over Tailscale). PUSH_KEY-gated; atomic write
        so a concurrent reader never sees a torn file. Returns (ok, msg, http_code)."""
        if not PUSH_KEY:               return False, "push disabled (no PUSH_KEY on this node)", 403
        if key != PUSH_KEY:            return False, "bad key", 403
        if name not in _PUSH_TARGETS:  return False, f"unknown log '{name}'", 400
        try:
            n = int(self.headers.get("Content-Length", 0))
        except ValueError:
            n = 0
        if n <= 0 or n > 64 * 1024 * 1024:   # heard is ~0.4MB; 64MB is a generous OOM guard
            return False, "empty or oversized body", 400
        body, dst = self.rfile.read(n), _PUSH_TARGETS[name]
        try:
            os.makedirs(os.path.dirname(dst), exist_ok=True)
            tmp = dst + ".tmp"
            with open(tmp, "wb") as f:
                f.write(body)
            os.replace(tmp, dst)         # atomic
        except OSError as e:
            return False, f"write failed: {e}", 500
        return True, f"wrote {len(body)}B -> {name}", 200

    def log_message(self, *a):
        pass   # quiet — systemd journal would otherwise log every static GET


def main():
    # clear any NOAA pipeline orphaned by a previous serve.py (start_new_session survives our exit).
    # try/except: pkill is absent in the NAS slim container (presentation role — no rtl_fm to kill there);
    # on the VM pkill exists and this runs exactly as before.
    try:
        subprocess.run(["pkill", "-f", f"rtl_fm.*-f {NOAA_FREQ}"], stderr=subprocess.DEVNULL)
    except FileNotFoundError:
        pass
    # Keep the voice-activity log (heard.jsonl) current from captured clips on a BACKGROUND timer — so the
    # correlate graph still gets the voice signal without the website globbing the clip dir on every
    # /api/heard poll. On the NAS (no clips) it's a cheap no-op; on the VM it logs new captures. Wrapped so
    # it can never crash the server. (Voice capture/save continues as before; we just don't scan it live.)
    def _ingest_loop():
        while True:
            try: _ingest_heard()
            except Exception: pass
            time.sleep(60)
    threading.Thread(target=_ingest_loop, daemon=True).start()
    atexit.register(_stop)
    httpd = ThreadingHTTPServer((BIND, PORT), Handler)
    # BIND_EXTRA (VM only): a second listener on the tailnet IP so (a) the NAS can forward control
    # POSTs here and (b) the owner's own tailnet devices reach the full-control dash directly. Localhost
    # stays the primary; the extra listener shares the same Handler/state.
    extra = os.environ.get("BIND_EXTRA", "")
    if extra:
        try:
            httpd2 = ThreadingHTTPServer((extra, PORT), Handler)
            threading.Thread(target=httpd2.serve_forever, daemon=True).start()
            print(f"istrain dashboard ALSO on http://{extra}:{PORT} (tailnet)")
        except OSError as e:
            print(f"BIND_EXTRA {extra} failed: {e} — continuing on {BIND} only")
    print(f"istrain dashboard + control on http://{BIND}:{PORT}  (per-radio NOAA toggle)")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        _stop()


if __name__ == "__main__":
    main()