#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Waveline remote-access relay (server side, runs on your-domain).

Bridges a Waveline desktop app (anywhere) and a phone browser (anywhere):

  app   ->  POST /app/sync   (token-authed; posts player state, drains commands)
  phone ->  POST /c/<TH>/cmd   (whitelisted media actions only; TH = the
  phone ->  GET  /c/<TH>/state     session's token hash, from /account/remote)

SECURITY MODEL (design constraint: nobody hacks into it):
  * ACCOUNT-ONLY identity (2026-09-27): the app's 256-bit DPAPI token is
    the whole credential - stored only as a SHA-256 hash, never kept raw;
    revocable by the app at any time.  There are NO pairing codes; the
    only user-facing code is the dashboard's 15-minute account LINK code.
  * No persistence: everything lives in memory; a reboot wipes all pairings.
  * Enumerability: unknown / unpaired / expired refs all return the SAME
    generic 404 body - no oracle for guessing sessions.
  * Commands: strict whitelist of media actions; body size caps; the relay
    never executes anything, it only matches strings between two clients.
    A fully compromised relay can do no more than play/pause/next.
  * Unprivileged deployment: runs as a non-root systemd service on
    127.0.0.1:<port>; TLS terminates at the existing nginx/caddy.
  * Rate limits: sliding-window per-IP buckets on every endpoint.

Standard library only - no third-party code to attack through.
"""

import base64
import collections
import hashlib
import hmac
import ipaddress
import json
import os
import re
import secrets
import threading
import time
from collections import defaultdict, deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs

# Dual-mode import: as a package (tests) or standalone (the VPS runs
# this file directly, with its siblings copied alongside).
try:
    from . import liveroom
except ImportError:  # pragma: no cover - standalone deployment
    import os as _os
    import sys as _sys
    _sys.path.insert(0, _os.path.dirname(_os.path.abspath(__file__)))
    import liveroom
try:
    from . import accounts as _accounts
except ImportError:  # pragma: no cover - standalone deployment
    import os as _os
    import sys as _sys
    _sys.path.insert(0, _os.path.dirname(_os.path.abspath(__file__)))
    import accounts as _accounts
try:
    from . import vpsagent as _vpsagent
except ImportError:  # pragma: no cover - standalone deployment
    import vpsagent as _vpsagent


MAX_BODY = 4096
# /app/agent/cookies/put alone needs a bigger body: the payload wraps a
# whole cookies.txt (27+ YouTube rows with long tokens) in JSON, and a
# REAL export lands around 5-6 KB - well over MAX_BODY.  The generic cap
# silently discarded such bodies and _reject() answered the misleading
# 'not found' (owner report 2026-10-02).  The worker's own store cap
# (vpsagent.COOKIES_MAX_BYTES, 512 KB) stays the real limit; this is
# only the transport ceiling, and ONLY for that one route.
COOKIES_BODY_MAX = 64 * 1024
# Live Room relay-side caps: a byte pipe, nothing more.
LR_ROOM_TTL = 5 * 60        # room dies 5 min after the last push
LR_CHUNK_MAX = 131072       # one push bigger than this is refused
LR_LISTENERS = 20           # concurrent listeners per room
LR_ROOMS = 8                # concurrent rooms server-wide

# Listener page: dark, big targets, screen-reader friendly.  No autoplay
# (browsers block it anyway) - an explicit Play button starts the stream;
# the status line only changes when the text changes (never re-announces
# the same thing to screen readers).
LR_PAGE = """<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Waveline Live Room</title>
<style>
body{background:#14171c;color:#e8e8e8;font-family:sans-serif;margin:0;
padding:2rem 1rem;text-align:center}
h1{font-size:1.4rem}
#status{min-height:1.5em;color:#9ab;margin:1rem 0}
button{font-size:1.2rem;padding:0.8rem 2rem;margin:0.3rem;border-radius:8px;
border:1px solid #3a4150;background:#1d222b;color:#e8e8e8}
audio{width:280px;margin-top:1rem}
</style></head><body>
<h1>Waveline Live Room</h1>
<div id="status" role="status" aria-live="polite">Loading…</div>
<button onclick="play()">Play</button>
<button onclick="stopa()">Stop</button>
<audio id="audio" controls preload="none"></audio>
<script>
var TOKEN="%(token)s";
var BASE="/lr/room/"+TOKEN;
var last="";
function set(s){if(s!==last){last=s;document.getElementById("status").textContent=s;}}
function poll(){fetch(BASE+"/meta").then(function(r){return r.json();}).then(function(j){
 if(!j.ok){set("This room has ended.");return;}
 var t=j.title?("Now playing: "+j.title+". "):"Live. ";
 set(t+j.listeners+(j.listeners===1?" person listening.":" people listening."));
}).catch(function(){set("Connection lost - retrying…");});}
setInterval(poll,4000);poll();
function play(){var a=document.getElementById("audio");a.src=BASE+"/stream";a.play();set("Connecting…");}
function stopa(){var a=document.getElementById("audio");a.pause();a.removeAttribute("src");set("Stopped.");}
</script></body></html>"""
# ACCOUNT-ONLY identity (2026-09-27): no pairing codes exist - the
# app's DPAPI-protected token is the whole credential and sessions are
# addressed by its HASH on the wire.
SESSION_IDLE_TTL = 30 * 24 * 3600  # pairing dies after 30 days without
# ANY activity - the app OR a remembered device polling (the cloud
# library must stay playable on the phone while the PC is off for
# days; was 24 h, which killed pairings during long PC-off periods)
TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{40,60}$")
# A /c/<ref> URL ref: the session's 64-hex token hash (a legacy 8-10
# char pairing code is still accepted by the ROUTES so old bookmarks
# fall through to the miss page instead of a bare 404).
PCREF_RE = r"([A-Z2-9]{8,10}|[0-9a-f]{64})"

# -- AudioVault-to-VPS downloads (2026-09-26) --------------------------------
# The PC's AudioVault dialog can ask the VPS to download a title into the
# OWNER'S Cloud Library (per linked-PC session lib), so it becomes
# playable on the phone/dashboard while the PC is off.  The job queue is
# in-memory (restart drops queued jobs - acceptable: the PC shows the
# final status and can re-ask), credentials NEVER reach the server: the
# worker logs in with a session cookie like a browser, holds nothing.
AV_WORKER_INTERVAL = 4.0        # seconds between queue polls
AV_FETCH_TIMEOUT = 60          # per-request timeout for the VPS-side client
AV_CONSECUTIVE_BLOCKS = 3      # challenge/4xx runs before backoff pauses
AV_BACKOFF = 15 * 60           # ... pause fetching this long (be polite)

# Only these actions may ever pass through the relay.  Mirrors the local
# web remote's command set; anything else is rejected outright.
VALID_ACTIONS = frozenset({
    "play_pause", "stop", "next_track", "prev_track",
    "seek_back", "seek_forward", "volume_up", "volume_down",
    "mute", "record_toggle", "record_now", "stop_recording",
    "toggle_shuffle", "cycle_repeat", "stop_after_current",
    "speed_up", "speed_down", "speed_reset",
    "bass_up", "bass_down", "treble_up", "treble_down", "tone_reset",
    "chapter_next", "chapter_prev", "bookmark_add", "announce_time",
    "sleep_timer", "cloud_upload_now", "cloud_delete",
    # 7.1: the alarm rings on the PC - silence it from the phone in bed
    # (snooze = stop now, the same alarm starts again after its snooze
    # minutes; dismiss = silence it for good).
    "alarm_snooze", "alarm_dismiss",
})

STATE_KEYS = {"now_playing", "playing", "volume", "position",
              "url", "live", "alarm_active", "alarm_label"}
MAX_STR = 200
MAX_URL = 500
MAX_BLOB = 3072        # /app/settings blob cap.  MUST stay below MAX_BODY
                       # (4096): the generic _body() reader refuses bigger
                       # requests before the handler sees them.  3 KB holds
                       # the ~30 preference keys several times over;
                       # accounts.SETTINGS_MAX_BYTES re-checks post-clean.
AV_MAX_TITLE = 200     # AudioVault job title cap (matches MAX_STR style)
AV_MAX_URL = 500       # AudioVault job URL cap (matches MAX_URL)
AV_HISTORY_MAX = 50    # rows kept in the persistent AudioVault history file

# Security-alert mail queue: module-level (created once at import, so
# per-request init races cannot orphan a pump); the pump thread itself
# is started lazily by the first request (idempotent).  DRAIN_INTERVAL
# is how often recorded-but-unmailed events (verify/reset/link/unlink)
# are swept into the queue - tests shrink it to near-zero.
_ALERTQ = collections.deque(maxlen=64)
_ALERT_SIG = threading.Semaphore(0)
DRAIN_INTERVAL = 60.0


# 8..32: AV uploads generate 24-char ids, the VPS agent prefixes
# "agent" + 20 hex (25) - entry 85: the cap must cover BOTH or
# agent recordings can never be deleted or streamed.  One shared
# compiled pattern: _Store and _Handler MUST agree (the AV worker
# validates through _Store.LIB_RE).
_LIB_RE = re.compile(r"^[A-Za-z0-9_-]{8,32}$")


class _Rate:
    """Per-IP sliding-window limiter (thread-safe, self-pruning)."""

    def __init__(self, per_minute: int, burst: int):
        self.per_minute = per_minute
        self.burst = burst
        self._hits = defaultdict(deque)
        self._lock = threading.Lock()

    def allow(self, ip: str) -> bool:
        now = time.monotonic()
        with self._lock:
            dq = self._hits[ip]
            while dq and now - dq[0] > 60.0:
                dq.popleft()
            if len(dq) >= self.burst:
                return False
            # sustained rate check
            recent = sum(1 for t in dq if now - t <= 60.0)
            if recent >= self.per_minute:
                return False
            dq.append(now)
            # prune stale ips occasionally
            if len(self._hits) > 4096:
                stale = [k for k, v in self._hits.items()
                         if not v or now - v[-1] > 3600.0]
                for k in stale:
                    del self._hits[k]
            return True


def _effective_ip(peer: str, forwarded: str) -> str:
    """Client IP for rate limiting, hardened against XFF forgery.

    Only a loopback TCP peer may speak for a proxy (our Apache), and even
    then we take the LAST X-Forwarded-For entry validated as an IP - the
    entry Apache itself appended.  A direct client's forged XFF is
    ignored entirely.  Mirrors relay_server.py's policy."""
    try:
        if ipaddress.ip_address(peer).is_loopback and forwarded:
            parts = [p.strip() for p in forwarded.split(",") if p.strip()]
            for part in reversed(parts):
                try:
                    ipaddress.ip_address(part)
                    return part
                except ValueError:
                    continue
    except ValueError:
        pass
    return peer


class _Abuse:
    """fail2ban-style daily abuse summary for the admin page.

    Counts every throttled (429), auth-rejected (404) and forbidden
    (403) response, per UTC day, with a bounded per-IP breakdown so the
    operator can see WHO is hammering the relay without tailing logs.

    Memory-bounded by design (an attacker must not be able to grow this
    table to hurt the relay): at most _MAX_IPS distinct IPs are tracked
    per day (overflowing IPs still count in the day totals), and days
    older than _DAYS are pruned on every write.
    """

    _DAYS = 8
    _MAX_IPS = 256
    _KINDS = ("429", "404", "403")

    def __init__(self):
        self._lock = threading.Lock()
        self._totals = {}   # "YYYY-MM-DD" -> {kind: count}
        self._ips = {}      # "YYYY-MM-DD" -> {ip: {kind: count}}

    @staticmethod
    def _today() -> str:
        return time.strftime("%Y-%m-%d", time.gmtime())

    def record(self, kind: str, ip: str) -> None:
        if kind not in self._KINDS or not ip:
            return
        day = self._today()
        with self._lock:
            self._prune_locked(day)
            totals = self._totals.setdefault(day, {k: 0 for k in self._KINDS})
            totals[kind] = totals.get(kind, 0) + 1
            ips = self._ips.setdefault(day, {})
            if ip in ips or len(ips) < self._MAX_IPS:
                per = ips.setdefault(ip, {k: 0 for k in self._KINDS})
                per[kind] = per.get(kind, 0) + 1

    def _prune_locked(self, today: str) -> None:
        # ISO dates sort lexically.  Keep `today` and the newest
        # _DAYS-1 earlier days; anything older is dropped.
        cutoff = today if len(self._totals) <= self._DAYS else \
            sorted(d for d in self._totals if d < today)[- (self._DAYS - 1)]
        for d in [d for d in self._totals if d < cutoff]:
            del self._totals[d]
            self._ips.pop(d, None)

    def summary(self) -> dict:
        with self._lock:
            self._prune_locked(self._today())
            days = []
            for day in sorted(self._totals, reverse=True):
                totals = self._totals[day]
                ips = self._ips.get(day, {})
                top = sorted(
                    ips.items(),
                    key=lambda kv: sum(kv[1].values()),
                    reverse=True)[:5]
                days.append({
                    "date": day,
                    "rate_limited": totals.get("429", 0),
                    "auth_failures": totals.get("404", 0),
                    "forbidden": totals.get("403", 0),
                    "top_ips": [
                        {"ip": ip,
                         "rate_limited": per.get("429", 0),
                         "auth_failures": per.get("404", 0),
                         "forbidden": per.get("403", 0)}
                        for ip, per in top],
                })
            return {"days": days,
                    "tracked_ips": sum(len(v) for v in self._ips.values())}


class _Store:
    LIB_RE = _LIB_RE
    TH_RE = re.compile(r"^[0-9a-f]{64}$")  # token-hash refs (account-only)
    """Pairing sessions, keyed by the app's token HASH.

    ACCOUNT-ONLY identity (2026-09-27): there are no pairing codes and
    no invitations - a phone control device registers by its device id
    and the owner bans what they do not recognise.

    PERSISTENCE (2026-09-21, the user: "I don't want to enter codes every
    time"): sessions survive relay restarts via a JSON file containing
    HASHES ONLY - the raw tokens never exist server-side, so a stolen
    db file is useless for impersonating anyone (the token lives only in
    the user's app).  A restarting Waveline re-registers with its token
    and the relay finds its session again; already-paired phones re-attach
    through their saved device id.  Idle TTL still applies on top.
    """

    def __init__(self, persist: bool = True):
        self._lock = threading.Lock()
        self._persist = persist
        self._pairs = {}  # token_hash -> session record
        self._acct_extras = {}  # email -> {bookmarks, listen, juke, digest}
        self._path = os.environ.get(
            "WAVELINE_RELAY_DB",
            os.path.join(os.path.expanduser("~"), ".waveline-relay",
                         "pairs.json"))
        if persist:
            self._load()

    # -- disk -------------------------------------------------------------
    def _load(self):
        try:
            with open(self._path, "r", encoding="utf-8") as f:
                rows = json.load(f)
            now = time.time()
            for row in rows:
                if now - row.get("last_seen", 0) > SESSION_IDLE_TTL:
                    continue
                rec = {
                    "state": {}, "commands": [],
                    "last_seen": float(row.get("last_seen", now)),
                    "created": float(row.get("created", now)),
                    "devices": {}, "consumed": False,
                    "lib": {},
                }
                for d in row.get("devices", []):
                    rec["devices"][str(d.get("id", ""))[:64]] = {
                        "id": str(d.get("id", ""))[:64],
                        "name": str(d.get("name", ""))[:40],
                        "last_seen": float(d.get("last_seen", now)),
                        "first_seen": float(d.get("first_seen", now)),
                        "banned": bool(d.get("banned", False)),
                    }
                rec["name"] = str(row.get("name", ""))[:40]
                for lid, lrow in row.get("lib", {}).items():
                    if not isinstance(lrow, dict):
                        continue
                    rec["lib"][str(lid)[:32]] = {
                        "id": str(lid)[:32],
                        "name": str(lrow.get("name", ""))[:MAX_STR],
                        "added": float(lrow.get("added", now)),
                        "bytes": max(0, int(lrow.get("bytes", 0))),
                        "file": str(lrow.get("file", ""))[:120],
                    }
                self._pairs[str(row.get("th", ""))] = rec
            # The extras row is a dict, not a list of sessions.
            for row in rows:
                if isinstance(row, dict) and "acct_extras" in row \
                        and isinstance(row["acct_extras"], dict):
                    self._acct_extras = row["acct_extras"]
        except FileNotFoundError:
            pass
        except Exception:
            # A corrupt db must never keep the relay down: start empty.
            try:
                os.replace(self._path, self._path + ".corrupt")
            except Exception:
                pass

    def _save_locked(self):
        if not self._persist:
            return
        rows = []
        for th, rec in self._pairs.items():
            rows.append({
                "th": th,
                "last_seen": rec["last_seen"],
                "created": rec["created"],
                "devices": [
                    {"id": d["id"], "name": d["name"],
                     "last_seen": d["last_seen"], "first_seen": d["first_seen"],
                     "banned": d["banned"]}
                    for d in rec["devices"].values()],
                "lib": {lid: {"name": l["name"], "added": l["added"],
                              "bytes": l["bytes"], "file": l["file"]}
                        for lid, l in rec.get("lib", {}).items()},
                "name": rec.get("name", ""),
            })
        # Account-wide extras (bookmarks / listen stats / jukebox) ride
        # the same file - emails and hashes only, never tokens.
        try:
            rows.append({"acct_extras": self._acct_extras})
        except Exception:
            pass
        try:
            os.makedirs(os.path.dirname(self._path), exist_ok=True)
            tmp = self._path + ".tmp"
            with open(tmp, "w", encoding="utf-8") as f:
                json.dump(rows, f)
            os.replace(tmp, self._path)  # atomic on POSIX and Windows
        except Exception:
            pass  # persistence is best-effort; the relay must not die

    @staticmethod
    def _hash(token: str) -> str:
        return hashlib.sha256(token.encode("utf-8")).hexdigest()

    # -- account-wide extras (2026-09-27, entry 49): bookmarks, listen
    # stats and jukebox requests.  Keyed by ACCOUNT EMAIL so every linked
    # PC shares them; tiny dicts ride the same pairs.json persistence.
    BOOKMARK_CAP = 200          # per account, newest kept
    STAT_TOP = 10               # digest top-N

    def _acct_rec(self, email: str) -> dict:
        """The per-account extras record (created on demand)."""
        email = str(email or "").lower()
        rec = self._acct_extras.get(email)
        if rec is None:
            rec = {"bookmarks": {}, "listen": {}, "juke": None,
                   "digest": 0.0}
            self._acct_extras[email] = rec
        return rec

    def bookmark_put(self, email: str, bid: str, url: str, name: str,
                     position: float, label: str) -> None:
        with self._lock:
            bms = self._acct_rec(email).setdefault("bookmarks", {})
            bms[bid] = {"url": str(url)[:MAX_URL],
                        "name": str(name)[:MAX_STR],
                        "pos": max(0.0, float(position)),
                        "label": str(label or "")[:80],
                        "at": time.time()}
            if len(bms) > self.BOOKMARK_CAP:
                for k in sorted(bms, key=lambda k: bms[k]["at"]) \
                        [:len(bms) - self.BOOKMARK_CAP]:
                    bms.pop(k, None)
            self._save_locked()

    def bookmark_del(self, email: str, bid: str) -> bool:
        with self._lock:
            bms = self._acct_rec(email).get("bookmarks", {})
            if bms.pop(str(bid)[:64], None) is None:
                return False
            self._save_locked()
            return True

    def bookmarks(self, email: str) -> list:
        with self._lock:
            bms = self._acct_rec(email).get("bookmarks", {})
            rows = [(v["at"], k, v) for k, v in bms.items()]
            rows.sort(key=lambda r: -r[0])
            return [(k, dict(v)) for _at, k, v in rows]

    # -- cross-device playback positions (6.0, "continue anywhere").
    # Keyed by ACCOUNT; one record per track with device + timestamp, so
    # any linked PC can offer to resume where another left off.
    POS_CAP = 300              # per account, most recent kept

    def position_put(self, email: str, track: str, pos: float,
                     device: str) -> None:
        with self._lock:
            rec = self._acct_rec(email).setdefault("positions", {})
            rec[str(track)[:MAX_URL]] = {
                "pos": max(0.0, float(pos)),
                "device": str(device or "")[:40],
                "at": time.time()}
            if len(rec) > self.POS_CAP:
                for k in sorted(rec, key=lambda k: rec[k]["at"])                         [:len(rec) - self.POS_CAP]:
                    rec.pop(k, None)
            self._save_locked()

    def positions(self, email: str) -> list:
        with self._lock:
            rec = self._acct_rec(email).get("positions", {})
            return [(k, dict(v)) for k, v in rec.items()]

    def listen_add(self, email: str, name: str, seconds: float) -> None:
        """Accumulate listening seconds per track (digest input)."""
        name = str(name or "")[:MAX_STR]
        if not name or seconds <= 0:
            return
        with self._lock:
            listen = self._acct_rec(email).setdefault("listen", {})
            listen[name] = round(listen.get(name, 0.0) + float(seconds), 1)
            self._save_locked()

    def listen_top(self, email: str) -> list:
        with self._lock:
            listen = dict(self._acct_rec(email).get("listen", {}))
        top = sorted(listen.items(), key=lambda kv: -kv[1])[:self.STAT_TOP]
        return [(n, round(s, 1)) for n, s in top]

    def listen_total(self, email: str) -> float:
        with self._lock:
            return round(sum(self._acct_rec(email).get(
                "listen", {}).values()), 1)

    def digest_candidates(self, week: float) -> list:
        """Accounts whose accumulator last rolled over a week ago AND
        that actually listened since (no empty-mail spam)."""
        now = time.time()
        out = []
        with self._lock:
            for email, rec in self._acct_extras.items():
                if now - float(rec.get("digest", 0.0)) < week:
                    continue
                if rec.get("listen"):
                    out.append(email)
        return out

    def listen_reset(self, email: str) -> None:
        """Clear the accumulator after a digest is mailed (weekly roll)."""
        with self._lock:
            self._acct_rec(email)["listen"] = {}
            self._acct_rec(email)["digest"] = time.time()
            self._save_locked()

    def listen_last_digest(self, email: str) -> float:
        with self._lock:
            return float(self._acct_rec(email).get("digest", 0.0))

    # -- jukebox: one live share per account ----------------------------
    def juke_create(self, email: str) -> dict:
        """(Re)issue the account's share token; returns the record."""
        with self._lock:
            rec = self._acct_rec(email)
            tok = secrets.token_urlsafe(24)
            rec["juke"] = {"tok": tok, "open": True, "req": {},
                           "now": {}, "reqs_served": 0}
            self._save_locked()
            return dict(rec["juke"])

    def juke_revoke(self, email: str) -> None:
        with self._lock:
            self._acct_rec(email)["juke"] = None
            self._save_locked()

    def juke_get(self, email: str):
        with self._lock:
            rec = self._acct_rec(email).get("juke")
            return dict(rec) if rec else None

    def juke_by_token(self, jtok: str):
        """(email, juke-record) for a live share token, or (None, None)."""
        jtok = str(jtok or "")
        with self._lock:
            for email, rec in self._acct_extras.items():
                j = rec.get("juke")
                if j and hmac.compare_digest(str(j.get("tok", "")), jtok):
                    return email, dict(j)
        return None, None

    JUKE_REQ_CAP = 50          # bounded pending queue per account

    def juke_request(self, email: str, rid: str, lid: str, name: str,
                     guest: str) -> str:
        """Queue a request; returns 'ok', 'closed' or 'dupe'."""
        with self._lock:
            j = self._acct_rec(email).get("juke")
            if not j or not j.get("open"):
                return "closed"
            if any(r.get("lid") == lid for r in j["req"].values()
                   if not r.get("done")):
                return "dupe"
            if len([r for r in j["req"].values() if not r.get("done")]) \
                    >= self.JUKE_REQ_CAP:
                return "closed"
            j["req"][str(rid)[:40]] = {"lid": str(lid)[:40],
                                       "name": str(name)[:MAX_STR],
                                       "guest": str(guest)[:24],
                                       "at": time.time(), "done": False}
            self._save_locked()
            return "ok"

    def juke_requests(self, email: str) -> list:
        with self._lock:
            j = self._acct_rec(email).get("juke")
            if not j:
                return []
            rows = [(v["at"], k, dict(v)) for k, v in j["req"].items()]
            rows.sort(key=lambda r: r[0])
            return [(k, v) for _at, k, v in rows]

    def juke_decide(self, email: str, rid: str, approve: bool) -> dict:
        """Approve/deny one request.  Returns {} if unknown; otherwise
        the request row (with 'done' set) - on approval it also carries
        the library lid for playback and bumps the session's queue."""
        rid = str(rid)[:40]
        with self._lock:
            j = self._acct_rec(email).get("juke")
            if not j or rid not in j["req"]:
                return {}
            row = j["req"][rid]
            if row.get("done"):
                return dict(row)
            row["done"] = True
            row["approved"] = bool(approve)
            if approve:
                j["reqs_served"] = int(j.get("reqs_served", 0)) + 1
                # push onto the owner's PC queue via push_by_th below
            self._save_locked()
            return dict(row)

    def juke_set_now(self, email: str, info: dict) -> None:
        """What the shared player is currently serving (guest page)."""
        with self._lock:
            j = self._acct_rec(email).get("juke")
            if j is not None:
                j["now"] = {k: str(v)[:MAX_STR] for k, v in
                            (info or {}).items() if v}
                self._save_locked()

    def juke_set_open(self, email: str, open_: bool) -> None:
        with self._lock:
            j = self._acct_rec(email).get("juke")
            if j is not None:
                j["open"] = bool(open_)
                self._save_locked()


    def find_by_th(self, th: str):
        """Live session record by token hash (or None)."""
        with self._lock:
            return self._pairs.get(str(th)[:64])

    def pair(self, token: str, name: str = "") -> bool:
        """ACCOUNT-ONLY identity (owner order 2026-09-27): the token IS
        the session.  No codes are stored, announced or rotated."""
        th = self._hash(token)
        now = time.time()
        if name:
            name = name.strip()[:40]
        with self._lock:
            rec = self._pairs.get(th)
            if rec is None:
                rec = {"state": {}, "commands": [],
                       "last_seen": now, "created": now, "devices": {},
                       "consumed": False, "lib": {}, "name": ""}
                self._pairs[th] = rec
            if name:
                rec["name"] = name  # newest choice wins; empty keeps old
            rec["last_seen"] = now
            self._save_locked()
            return True

    def auth(self, token: str) -> bool:
        """Is the session for *token* live?  (Token-only identity.)"""
        with self._lock:
            rec = self._pairs.get(self._hash(token))
            if rec is None:
                return False
            if time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                del self._pairs[self._hash(token)]
                return False
            return True

    def th_for(self, ref: str) -> str:
        """Hashed token of the session for *ref* ('' = none).  *ref* may
        be the raw token (hashed here; the raw token is never kept) or
        already a 64-hex token hash (used as-is).  Callers use the hash
        for library calls."""
        with self._lock:
            ref = str(ref or "")
            th = ref if self.TH_RE.match(ref) else self._hash(ref)
            if th not in self._pairs:
                return ""
            return th

    def th_for_code(self, code: str) -> str:
        """Retired name; kept for callers - resolves a TOKEN HASH ref
        (hashes and raw tokens hash to the same value; a live session
        is required either way)."""
        with self._lock:
            ref = str(code or "")
            th = ref if self.TH_RE.match(ref) else self._hash(ref)
            rec = self._pairs.get(th)
            if rec is None:
                return ""
            if time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return ""
            return th

    def device_ok(self, code: str, devid: str) -> bool:
        """True only if *devid* is an ALREADY-REGISTERED, non-banned
        device of the session addressed by TOKEN HASH.  Never registers
        anything (read endpoints use this)."""
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None or time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return False
            dev = rec["devices"].get(str(devid)[:64])
            return dev is not None and not dev["banned"]

    def sync(self, th: str, state: dict):
        """App round trip by TOKEN HASH: update state, drain commands,
        report paired control devices.  (Codes were retired 2026-09-27.)"""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return [], []
            rec["state"] = state
            rec["last_seen"] = time.time()
            out, rec["commands"] = rec["commands"], []
            now = time.time()
            devices = [
                {"id": d["id"], "name": d["name"],
                 "idle_min": int((now - d["last_seen"]) // 60)}
                for d in rec["devices"].values() if not d["banned"]]
            return out, devices

    def device_seen(self, code: str, devid: str, name: str) -> str:
        """Register/update a control device.  *code* is the session's
        TOKEN HASH (account-only identity).  Returns 'ok', 'banned' or
        'missing'."""
        devid = str(devid or "")[:64]
        now = time.time()
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None or time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return "missing"
            dev = rec["devices"].get(devid)
            if dev is not None and dev["banned"]:
                return "banned"
            if dev is None:
                if len(rec["devices"]) >= 16:  # bounded table per session
                    live = [d for d in rec["devices"].values()
                            if not d["banned"]]
                    if live:
                        oldest = min(live, key=lambda d: d["last_seen"])
                        rec["devices"].pop(oldest["id"], None)
                dev = {"id": devid, "name": "", "last_seen": now,
                       "first_seen": now, "banned": False}
                rec["devices"][devid] = dev
                rec["consumed"] = True
                rec["just_paired"] = devid   # one-shot, drained by callers
                self._save_locked()
            dev["last_seen"] = now
            clean = str(name or "").strip()[:40]
            if clean and not dev.get("pinned"):
                # The owner can pin a name from the account dashboard;
                # pinned names are not overwritten by the device.
                dev["name"] = clean
            return "ok"

    def rename_device(self, th: str, devid: str, name: str) -> str:
        """Owner renames a control device from the dashboard.  The name
        is pinned (the device can no longer overwrite it).  Returns
        'ok' or 'missing'."""
        clean = str(name or "").strip()[:40]
        if not clean:
            return "missing"
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return "missing"
            dev = rec["devices"].get(str(devid)[:64])
            if dev is None or dev["banned"]:
                return "missing"
            dev["name"] = clean
            dev["pinned"] = True
            self._save_locked()
            return "ok"

    def ban_device_by_th(self, th: str, devid: str) -> int:
        """Owner-side revocation on a linked pairing (dashboard/in-app).
        *devid* "*" bans every device of the session.  Returns how many
        were banned."""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return 0
            if devid == "*":
                n = 0
                for d in rec["devices"].values():
                    if not d["banned"]:
                        d["banned"] = True
                        n += 1
                if n:
                    self._save_locked()
                return n
            dev = rec["devices"].get(str(devid)[:64])
            if dev is None or dev["banned"]:
                return 0
            dev["banned"] = True
            self._save_locked()
            return 1

    def devices_list(self, code: str) -> list:
        """Control devices for the session addressed by TOKEN HASH."""
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None:
                return []
            now = time.time()
            return [
                {"id": d["id"], "name": d["name"],
                 "idle_min": int((now - d["last_seen"]) // 60)}
                for d in rec["devices"].values() if not d["banned"]]

    def ban_device(self, code: str, devid: str) -> int:
        """Owner-side device revocation (TOKEN HASH addressing; kept as
        the in-app ban path alongside ban_device_by_th).  Returns how
        many were banned."""
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None:
                return 0
            if devid == "*":
                n = 0
                for d in rec["devices"].values():
                    if not d["banned"]:
                        d["banned"] = True
                        n += 1
                return n
            dev = rec["devices"].get(str(devid)[:64])
            if dev is None or dev["banned"]:
                return 0
            dev["banned"] = True
            self._save_locked()
            return 1

    def push(self, code: str, action: str, devid: str = "",
             name: str = "") -> str:
        """Queue a command.  *code* is the session's TOKEN HASH.
        Returns 'ok', 'banned' (device revoked) or 'missing'."""
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None or time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return "missing"
            if devid:
                dev = rec["devices"].get(str(devid)[:64])
                if dev is not None and dev["banned"]:
                    return "banned"
            # A token-hash ref IS the credential (account-only identity):
            # it may always queue; device tracking happens via headers.
            rec["commands"].append(action)
            del rec["commands"][:-32]
            return "ok"

    # -- cloud library (per session; survives relay restarts and PC-off) ---
    # SIZED PER USER REQUEST (2026-09-22, "max file size like 2 GB and
    # session like 20, I mostly use myself"): 2 GB per file, 20 GB per
    # session.  VPS has 112 GB free; the free-space check in lib_add
    # refuses writes before the disk itself is at risk.  Files stream to
    # disk in 64 KB chunks - size is bounded by quota, never by RAM.
    LIB_MAX_FILE = 2 * 1024 * 1024 * 1024     # 2 GB per file
    LIB_MAX_FILES = 200
    LIB_MAX_TOTAL = 20 * 1024 * 1024 * 1024   # 20 GB per session library
    LIB_MIN_FREE_DISK = 2 * 1024 * 1024 * 1024  # refuse at <2 GB disk free

    @staticmethod
    def _lib_dir() -> str:
        base = os.environ.get("WAVELINE_RELAY_LIB",
                              os.path.join(os.path.expanduser("~"),
                                           ".waveline-relay", "lib"))
        return base

    def _lib_file(self, th: str, lid: str) -> str:
        return os.path.join(self._lib_dir(),
                            "s" + th[:16], "%s.audio" % lid)

    @staticmethod
    def _free_disk_bytes(path: str) -> int:
        """Free bytes on the filesystem holding *path*; -1 if unknown.
        POSIX: os.statvfs.  Windows: ctypes GetDiskFreeSpaceExW (the
        relay runs on both - dev PC and VPS)."""
        if os.name == "nt":
            try:
                import ctypes
                free = ctypes.c_ulonglong(0)
                ok = ctypes.windll.kernel32.GetDiskFreeSpaceExW(
                    ctypes.c_wchar_p(path), None, None,
                    ctypes.byref(free))
                return free.value if ok else -1
            except Exception:
                return -1
        try:
            st = os.statvfs(path)
            return st.f_bavail * st.f_frsize
        except OSError:
            return -1

    def lib_add(self, th: str, lid: str, name: str, size: int) -> str:
        """Register an upload slot; caller streams bytes to the file.
        Returns 'ok', 'full' (quota/size), 'disk' (server low on space)
        or 'missing'."""
        if size <= 0 or size > self.LIB_MAX_FILE:
            return "full"
        free = self._free_disk_bytes(self._lib_dir())
        if free >= 0 and free < self.LIB_MIN_FREE_DISK + size:
            return "disk"
        with self._lock:
            rec = self._pairs.get(th)
            if rec is None:
                return "missing"
            lib = rec.setdefault("lib", {})
            total = sum(l["bytes"] for l in lib.values())
            if len(lib) >= self.LIB_MAX_FILES or \
                    total + size > self.LIB_MAX_TOTAL:
                return "full"
            lib[lid] = {"id": lid, "name": str(name)[:MAX_STR],
                        "added": time.time(), "bytes": size,
                        "file": self._lib_file(th, lid)}
            self._save_locked()
            return "ok"

    def lib_del(self, th: str, lid: str) -> bool:
        with self._lock:
            rec = self._pairs.get(th)
            if rec is None:
                return False
            l = rec.get("lib", {}).pop(lid, None)
            self._save_locked()
        if l:
            try:
                os.remove(l["file"])
            except OSError:
                pass
        return l is not None

    def lib_list(self, th: str) -> list:
        with self._lock:
            rec = self._pairs.get(th)
            if rec is None:
                return []
            return sorted(rec.get("lib", {}).values(),
                          key=lambda l: -l["added"])

    def lib_get(self, th: str, lid: str):
        with self._lock:
            rec = self._pairs.get(th)
            if rec is None:
                return None
            return rec.get("lib", {}).get(lid)

    def pop_just_paired(self, th: str) -> str:
        """Device id registered by the MOST RECENT device_seen on this
        session ('' = none).  Consumed on read so repeat connections
        never re-trigger (pair alerts, dashboard counters)."""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return ""
            return rec.pop("just_paired", "")

    def auth_token(self, token: str):
        """App identity by token alone (like revoke): returns the token
        hash of a live pairing session, or None.  Used by account link/
        unlink where the invitation code may already have rotated."""
        th = self._hash(token)
        with self._lock:
            rec = self._pairs.get(th)
            if rec is None:
                return None
            if time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return None
            return th

    def paired(self, code: str) -> bool:
        """Is the session addressed by TOKEN HASH live?"""
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None:
                return False
            if time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return False
            return True

    def state(self, code: str) -> dict:
        """Latest player state for the session addressed by TOKEN HASH."""
        with self._lock:
            rec = self._pairs.get(str(code)[:64])
            if rec is None:
                return {}
            return dict(rec["state"])

    # -- account dashboard views (by token hash; no codes are exposed) -----
    def state_by_th(self, th: str) -> dict:
        """Latest player state for a linked pairing, for the account
        dashboard.  Empty dict when unknown/expired."""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return {}
            return dict(rec["state"])

    def devices_by_th(self, th: str) -> list:
        """Control devices for a linked pairing (online = seen <120 s)
        with idle minutes for the dashboard."""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return []
            now = time.time()
            return [{"id": d["id"], "name": d["name"],
                     "online": now - float(d.get("last_seen", 0)) < 120,
                     "idle_min": int((now - float(d.get("last_seen", 0))) // 60),
                     "last_seen": float(d.get("last_seen", 0)),
                     "banned": bool(d.get("banned", False))}
                    for d in rec["devices"].values()
                    if not d.get("banned")]

    def pc_online(self, th: str) -> bool:
        """A pairing counts online when the app synced recently
        (mirrors the 15 s poll window the app-side watcher uses)."""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None:
                return False
            return (time.time() - rec["last_seen"]) < 30

    def push_by_th(self, th: str, action: str) -> str:
        """Queue a VALID_ACTION for a linked pairing from the dashboard.
        Only reachable for actions already whitelisted by _h_cmd."""
        with self._lock:
            rec = self._pairs.get(str(th)[:64])
            if rec is None or \
                    time.time() - rec["last_seen"] > SESSION_IDLE_TTL:
                return "missing"
            rec["commands"].append(str(action)[:40])
            del rec["commands"][:-32]
            return "ok"

    def unlinked_token_hashes(self) -> set:
        """Token hashes that still have a live pairing session but are
        NOT linked to any account device slot.  Unused in v1 endpoints;
        kept for the operator's admin tooling."""
        return set(self._pairs.keys())

    def revoke(self, code: str, token: str) -> str:
        """Revoke a session by TOKEN alone (the code argument is
        accepted but ignored); returns its token hash ('' if unknown) so
        the caller can tear down anything owned by it (Live Rooms)."""
        with self._lock:
            th = self._hash(token)
            if th in self._pairs:
                del self._pairs[th]
                self._save_locked()
                return th
            return ""

    def revoke_token(self, token: str) -> str:
        """ACCOUNT-ONLY revocation by token alone.  Returns the token
        hash ('' if unknown) for Live-Room teardown."""
        with self._lock:
            th = self._hash(token)
            if th in self._pairs:
                del self._pairs[th]
                self._save_locked()
                return th
            return ""

    def revoke_admin(self, code: str):
        """Admin revocation by TOKEN HASH (the code argument is accepted
        for caller compatibility but ignored).  Kills the whole session;
        returns its token hash ('' if unknown)."""
        with self._lock:
            th = str(code or "")[:64]
            if th not in self._pairs:
                return ""
            del self._pairs[th]
            self._save_locked()
            return th

    def list_admin(self) -> list:
        """Operator listing.  ACCOUNT-ONLY identity (2026-09-27): rows
        carry the session's TOKEN HASH as "th" (no secrets - hashes only,
        never recoverable), the client's chosen nickname, an ONLINE flag
        and what it is playing.  Online clients first, newest activity
        first."""
        now = time.time()
        with self._lock:
            rows = []
            for th, rec in self._pairs.items():
                online = (now - rec["last_seen"]) <= 150
                rows.append({
                    "th": str(th)[:64],
                    "name": str(rec.get("name", ""))[:40],
                    "online": online,
                    "idle_min": int((now - rec["last_seen"]) // 60),
                    "now_playing": str(rec["state"].get("now_playing",
                                                        ""))[:80],
                    "playing": bool(rec["state"].get("playing", False)),
                })
            rows.sort(key=lambda r: (not r["online"], -r["idle_min"]))
            return rows

    def sweep(self) -> None:
        now = time.time()
        with self._lock:
            stale = [t for t, r in self._pairs.items()
                     if now - r["last_seen"] > SESSION_IDLE_TTL]
            for t in stale:
                del self._pairs[t]
            # prune idle devices from live sessions too
            for rec in self._pairs.values():
                dead = [i for i, d in rec["devices"].items()
                        if now - d["last_seen"] > SESSION_IDLE_TTL]
                for i in dead:
                    del rec["devices"][i]
            self._save_locked()


ROOM_HOST_TTL = 600        # host silent 10 min -> room closes
ROOM_MEMBER_TTL = 1800     # member silent 30 min -> dropped
MAX_ROOMS = 64
MAX_MEMBERS = 16
VOICE_HISTORY = 20         # clips kept per room
VOICE_MAX_BYTES = 1_400_000  # decoded pcm cap (~43 s @16 kHz)
VOICE_RATES = (8000, 16000, 24000, 48000)


class _Rooms:
    """Listen Together rooms: public lobby, host-driven playback sync,
    push-to-talk voice clips.  Public by design: anyone may list and join.
    Everything in-memory; swept when the host goes silent."""

    def __init__(self):
        self._lock = threading.Lock()
        self._rooms = {}
        # outboxes of deleted rooms survive one sweep cycle so members
        # still learn WHY the room vanished ("room-closed")
        self._orphan_outbox = {}

    @staticmethod
    def _iphash(ip: str) -> str:
        return hashlib.sha256(ip.encode("utf-8")).hexdigest()

    def list_public(self) -> list:
        with self._lock:
            return [{
                "id": r["id"],
                "name": r["name"],
                "host": (r["members"].get(r["host_id"], {})
                         ).get("nick", "?"),
                "listeners": len(r["members"]),
                "now_playing": str(r["state"].get("title", "")),
            } for r in self._rooms.values()]

    def create(self, name: str, nick: str, ip: str):
        name = str(name or "").strip()[:40] or "Listen Together"
        nick = str(nick or "").strip()[:24] or "Host"
        now = time.time()
        with self._lock:
            if len(self._rooms) >= MAX_ROOMS:
                return None
            rid = "r" + secrets.token_hex(8)
            mid = secrets.token_hex(6)
            self._rooms[rid] = {
                "id": rid, "name": name, "created": now,
                "host_id": mid,
                "members": {mid: {"id": mid, "nick": nick,
                                  "ip": self._iphash(ip),
                                  "admin": True,  # host moderates
                                  "last_seen": now}},
                "state": {}, "events": {}, "outbox": {},
                "bans": set(), "voice": [], "last_voice_seq": 0,
                "last_publish": now,
            }
            return rid, mid

    def join(self, rid: str, nick: str, ip: str):
        nick = str(nick or "").strip()[:24] or "Listener"
        now = time.time()
        with self._lock:
            r = self._rooms.get(rid)
            if r is None:
                return "gone"
            iph = self._iphash(ip)
            if iph in r["bans"]:
                return "banned"
            if len(r["members"]) >= MAX_MEMBERS:
                return "full"
            mid = secrets.token_hex(6)
            r["members"][mid] = {"id": mid, "nick": nick, "ip": iph,
                                 "admin": False, "last_seen": now}
            return rid, mid, r["name"]

    def _member(self, rid: str, mid: str):
        r = self._rooms.get(rid)
        if r is None:
            return None, None
        m = r["members"].get(mid)
        if m is None:
            return r, None
        if time.time() - m["last_seen"] > ROOM_MEMBER_TTL:
            del r["members"][mid]
            return r, None
        return r, m

    def _drain_locked(self, r, mid):
        """Pop this member's pending events (in-room or already removed)."""
        events = r["events"].pop(mid, [])
        events += r["outbox"].pop(mid, [])
        return events

    def poll(self, rid: str, mid: str):
        """Touch membership; return state, member list and drained events.
        Removed members still get their one final event delivery
        (kicked/banned/room-closed) before the 404."""
        now = time.time()
        with self._lock:
            r = self._rooms.get(rid)
            if r is None:
                ob = self._orphan_outbox.get(rid)
                if ob and mid in ob:
                    return {"state": {}, "members": [],
                            "events": ob.pop(mid)}
                return None
            m = r["members"].get(mid)
            if m is None:
                if mid in r["outbox"]:
                    return {"state": {}, "members": [],
                            "events": self._drain_locked(r, mid)}
                return None
            if now - m["last_seen"] > ROOM_MEMBER_TTL:
                del r["members"][mid]
                if mid in r["outbox"]:
                    return {"state": {}, "members": [],
                            "events": self._drain_locked(r, mid)}
                return None
            m["last_seen"] = now
            events = self._drain_locked(r, mid)
            members = [{"id": x["id"], "nick": x["nick"],
                        "host": x["id"] == r["host_id"],
                        "admin": bool(x.get("admin"))}
                       for x in r["members"].values()]
            members.sort(key=lambda x: (not x["host"], x["nick"]))
            return {"state": dict(r["state"]), "members": members,
                    "events": events}

    def publish(self, rid: str, mid: str, state: dict) -> bool:
        """Host only.  state keys: title, url, position, playing, live."""
        with self._lock:
            r, m = self._member(rid, mid)
            if r is None or m is None or mid != r["host_id"]:
                return False
            m["last_seen"] = time.time()
            r["last_publish"] = time.time()
            r["state"] = state
            return True

    def leave(self, rid: str, mid: str) -> bool:
        """True = room closed (host left)."""
        with self._lock:
            r, m = self._member(rid, mid)
            if r is None or m is None:
                return False
            if mid == r["host_id"]:
                for other_mid in r["members"]:
                    r["outbox"][other_mid] = ["room-closed"]
                r["members"].clear()
                del self._rooms[rid]
                self._orphan_outbox[rid] = r["outbox"]
                return True
            del r["members"][mid]
            return False

    def kick(self, rid: str, actor: str, target: str, ban: bool,
             force: bool = False) -> str:
        """Host or ADMIN removes a member - including other admins and
        even the HOST (an admin breaking the rules gets removed too;
        the user's explicit moderation model, 2026-09-21).

        *force* (relay admin key) bypasses the in-room ladder entirely.
        The only hard rule: nobody removes themselves (leave exists for
        that).  When the removed member IS the host, the room survives:
        the most senior remaining admin becomes host (otherwise the
        earliest remaining member) - state, members and voice live on.
        """
        with self._lock:
            r = self._rooms.get(rid)
            if r is None:
                return "gone"
            if not force:
                a = r["members"].get(actor)
                if a is None:
                    return "missing"
                if not (actor == r["host_id"] or a.get("admin")):
                    return "forbidden"  # plain members cannot moderate
            if target == actor:
                return "missing"  # use leave, not kick
            m = r["members"].get(target)
            if m is None:
                return "missing"
            if ban:
                r["bans"].add(m["ip"])
            r["outbox"][target] = ["banned" if ban else "kicked"]
            del r["members"][target]
            if target == r["host_id"]:
                # HOST SUCCESSION: senior admin first, else earliest member.
                pool = [x for x in r["members"].values() if x.get("admin")]
                if not pool:
                    pool = list(r["members"].values())
                if not pool:
                    # nobody left: close the room cleanly
                    r["members"].clear()
                    del self._rooms[rid]
                    self._orphan_outbox[rid] = r["outbox"]
                    return "ok"
                new_host = min(pool, key=lambda x: x["last_seen"])
                new_host["admin"] = True
                r["host_id"] = new_host["id"]
                for x in r["members"].values():
                    r["outbox"].setdefault(x["id"], []).append(
                        "new-host:%s" % new_host["nick"])
            return "ok"

    def promote(self, rid: str, actor: str, target: str,
                grant: bool) -> str:
        """Host-only: grant or revoke moderator rights on a member."""
        with self._lock:
            r = self._rooms.get(rid)
            if r is None:
                return "gone"
            if actor != r["host_id"]:
                return "forbidden"
            m = r["members"].get(target)
            if m is None:
                return "missing"
            if target == r["host_id"]:
                return "missing"
            m["admin"] = bool(grant)
            r["outbox"][target] = ["promoted" if grant else "demoted"]
            return "ok"

    def voice_post(self, rid: str, mid: str, rate: int,
                   pcm_b64: str) -> str:
        try:
            pcm = base64.b64decode(pcm_b64, validate=True)
        except Exception:
            return "bad"
        if len(pcm) == 0 or len(pcm) > VOICE_MAX_BYTES:
            return "bad"
        if rate not in VOICE_RATES:
            return "bad"
        with self._lock:
            r, m = self._member(rid, mid)
            if r is None or m is None:
                return "gone"
            r["last_voice_seq"] += 1
            r["voice"].append({"seq": r["last_voice_seq"],
                               "nick": m["nick"],
                               "rate": rate, "pcm": pcm_b64})
            del r["voice"][:-VOICE_HISTORY]
            return "ok"

    def voice_sync(self, rid: str, mid: str, since: int):
        with self._lock:
            r, m = self._member(rid, mid)
            if r is None or m is None:
                return None
            m["last_seen"] = time.time()
            clips = [{"seq": c["seq"], "nick": c["nick"],
                      "rate": c["rate"], "pcm": c["pcm"]}
                     for c in r["voice"] if c["seq"] > since]
            return clips, r["last_voice_seq"]

    def sweep(self) -> None:
        now = time.time()
        with self._lock:
            dead = [rid for rid, r in self._rooms.items()
                    if now - r["last_publish"] > ROOM_HOST_TTL
                    or not r["members"]]
            for rid in dead:
                if self._rooms[rid]["outbox"]:
                    self._orphan_outbox[rid] = self._rooms[rid]["outbox"]
                del self._rooms[rid]
            for rid in [k for k in self._orphan_outbox
                        if not self._orphan_outbox[k]]:
                del self._orphan_outbox[rid]
            for r in self._rooms.values():
                gone = [mid for mid, m in r["members"].items()
                        if now - m["last_seen"] > ROOM_MEMBER_TTL]
                for mid in gone:
                    del r["members"][mid]


class _AudioVaultWorker:
    """Downloads AudioVault titles on the VPS into the owner's Cloud
    Library (2026-09-26).

    Why server-side: the phone/dashboard can then play the title while
    the PC is off (lib files live on the VPS), and the download does not
    occupy the PC.  The owner's AudioVault password NEVER reaches the
    server: the PC enqueues (title URL + name), the worker logs in to
    audiovault.net with its own browser session (Cloudflare-style site:
    datacenter IPs can face challenges, hence the challenge backoff),
    streams the file straight into the session lib via lib_add - quota,
    disk guard and persistence are exactly the Cloud Library's.

    PERSISTENT HISTORY (2026-09-26, entry 42): every job is recorded to
    a small JSON file (AV_HISTORY_PATH) so the dashboard's "AudioVault
    downloads" section shows pending/done/failed/blocked jobs across
    relay restarts.  The queue itself stays in-memory; after a restart
    un-finished history rows are simply marked failed (the PC shows the
    final status and can re-send) - honest, not silently stuck.
    """

    def __init__(self, store: "_Store"):
        self._store = store
        self._lock = threading.Lock()
        self._jobs = {}          # th -> [job dicts, FIFO]
        self._blocked_until = 0.0  # challenge/4xx backoff deadline
        self._fetch_failures = 0
        self._history = []       # persistent rows, newest first
        self._history_loaded = False
        self._load_history()
        # The VPS-side AudioVault session client (stdlib port).  Built
        # lazily: audiovault.py imports cleanly on POSIX (urllib only).
        self._client = None
        threading.Thread(target=self._loop, name="waveline-av-worker",
                         daemon=True).start()

    # -- persistent history (dashboard "AudioVault downloads") -----------
    @staticmethod
    def _history_path() -> str:
        return os.environ.get(
            "WAVELINE_RELAY_AV_HISTORY",
            os.path.join(os.path.expanduser("~"), ".waveline-relay",
                         "audiovault-history.json"))

    def _load_history(self) -> None:
        if not self._history_loaded:
            self._history_loaded = True
            try:
                with open(self._history_path(), "r",
                          encoding="utf-8") as fh:
                    rows = json.load(fh)
                if isinstance(rows, list):
                    self._history = [r for r in rows
                                     if isinstance(r, dict)]
                    for row in self._history:
                        if row.get("status") == "pending":
                            row["status"] = "failed"
                            row["detail"] = (
                                "server restarted before the download "
                                "finished")
                            row["updated"] = time.time()
            except (FileNotFoundError, ValueError, OSError):
                self._history = []

    def _history_append(self, th: str, email: str, url: str, name: str,
                        status: str, detail: str = "",
                        requested: float = None) -> None:
        """Add one row and persist (best-effort - history write failures
        must never break the download path)."""
        row = {"url": str(url)[:AV_MAX_URL],
               "name": str(name)[:AV_MAX_TITLE],
               "th": str(th)[:64], "email": str(email)[:190],
               "status": status, "detail": str(detail)[:160],
               "requested": float(requested if requested is not None
                                  else time.time()),
               "updated": time.time()}
        with self._lock:
            self._history.insert(0, row)
            del self._history[AV_HISTORY_MAX:]
            self._save_history_locked()

    def _history_set_status(self, th: str, url: str, status: str,
                            detail: str = "") -> None:
        with self._lock:
            for row in self._history:
                if row.get("th") == th and row.get("url") == url \
                        and row.get("status") == "pending":
                    row["status"] = status
                    row["detail"] = str(detail)[:160]
                    row["updated"] = time.time()
            self._save_history_locked()

    def _save_history_locked(self) -> None:
        try:
            path = self._history_path()
            os.makedirs(os.path.dirname(path), exist_ok=True)
            tmp = path + ".tmp"
            with open(tmp, "w", encoding="utf-8") as fh:
                json.dump(self._history, fh)
            os.replace(tmp, path)
        except Exception:
            pass

    def history_for(self, emails) -> list:
        """History rows (newest first) for any of *emails*, capped."""
        if isinstance(emails, str):
            emails = [emails]
        wanted = {str(e) for e in emails}
        with self._lock:
            return [dict(r) for r in self._history
                    if r.get("email") in wanted][:AV_HISTORY_MAX]

    # -- queue API (called from request handlers) -----------------------
    def enqueue(self, th: str, email: str, url: str, name: str,
                requester: str) -> str:
        """Queue one download for the session *th*.  Returns 'ok',
        'quota' (job would exceed the lib quota estimate) or 'missing'
        (unknown session)."""
        if not th or len(self._store.lib_list(th)) + 1 > \
                self._store.LIB_MAX_FILES:
            return "quota" if th else "missing"
        job = {"url": str(url)[:AV_MAX_URL], "name": str(name)[:AV_MAX_TITLE],
               "requested_by": str(requester)[:64],
               "requested": time.time()}
        with self._lock:
            already = any(existing["url"] == job["url"]
                          for existing in self._jobs.get(th, []))
            if not already:
                self._jobs.setdefault(th, []).append(job)
        if already:
            return "ok"          # already queued - do not duplicate
        # Record for the dashboard AFTER the lock (persist is best-effort).
        self._history_append(th, email, job["url"], job["name"], "pending",
                             requested=job["requested"])
        return "ok"

    def status(self, th: str) -> dict:
        with self._lock:
            queued = len(self._jobs.get(th, []))
        # Queued jobs are counted; finished ones are IN the lib (the
        # listing shows them).  Single global state message per session.
        return {"queued": queued,
                "active": getattr(self, "_active", ""),
                "note": self._note_for(th)}

    def _note_for(self, th: str) -> str:
        """Human-readable status hint for the session (for the phone UI)."""
        with self._lock:
            queued = len(self._jobs.get(th, []))
        active = getattr(self, "_active", "")
        if active:
            return "downloading %s" % active
        if queued:
            return "%d job(s) queued" % queued
        if time.time() < self._blocked_until:
            return "paused (audiovault.net is challenging this server; "
            "retrying later)"
        return ""

    # -- worker loop -------------------------------------------------------
    def _loop(self):
        while True:
            time.sleep(AV_WORKER_INTERVAL)
            try:
                self._tick()
            except Exception:
                pass

    def _tick(self):
        if time.time() < self._blocked_until:
            return
        job = None
        th = ""
        with self._lock:
            for t, jobs in self._jobs.items():
                if jobs:
                    th = t
                    job = jobs.pop(0)
                    break
        if job is None:
            return
        self._active = job["name"]
        try:
            ok = self._fetch_into_lib(th, job)
            if ok:
                self._history_set_status(th, job["url"], "done")
            else:
                self._history_set_status(th, job["url"], "failed")
        except Exception:
            try:
                self._history_set_status(th, job["url"], "failed")
            except Exception:
                pass
        finally:
            self._active = ""

    def _client_for(self):
        if self._client is None:
            try:
                from audiovault import AudioVaultClient
            except ImportError:
                from waveline.audiovault import AudioVaultClient
            self._client = AudioVaultClient()
        return self._client

    def _fetch_into_lib(self, th: str, job: dict) -> bool:
        """Download job['url'] and register it in the session lib.
        Streams into the final lib file path after lib_add reserves the
        slot (same shape as the app's /app/lib/upload handler).  Returns
        True when the file landed completely."""
        import urllib.request as _ur
        client = self._client_for()
        url, name = job["url"], job["name"]
        try:
            req = _ur.Request(url)
            for k, v in AV_HEADERS.items():
                req.add_header(k, v)
            r = client._opener.open(req, timeout=AV_FETCH_TIMEOUT)
        except Exception:
            self._count_failure()
            return False
        with r:
            ctype = (r.headers.get("Content-Type") or "").lower()
            if ctype.startswith("text/html") or \
                    (r.status or 0) >= 400:
                # Login page / challenge / not a file: count the failure
                # (consecutive runs trip the backoff) and drop the job;
                # the history row shows failed.
                self._count_failure()
                return False
            total = int(r.headers.get("Content-Length") or 0)
            if total <= 0 or total > _Store.LIB_MAX_FILE:
                return False
            # Reserve the slot with the ESTIMATED size; correct after the
            # stream (Content-Length may lie with compression off... but
            # lib entries hold 'bytes' for the listing only).
            lid = "av" + secrets.token_urlsafe(12).replace("-", "A") \
                .replace("_", "B")
            lid = ("av" + secrets.token_hex(9))[:24]
            lid = re.sub(r"[^A-Za-z0-9_-]", "A", lid)
            if not _Store.LIB_RE.match(lid):
                lid = "av" + secrets.token_hex(8)
            result = self._store.lib_add(th, lid, name or "AudioVault title",
                                         total)
            if result != "ok":
                return False                # quota/disk - job dropped
            path = self._store._lib_file(th, lid)
            done = 0
            try:
                os.makedirs(os.path.dirname(path), exist_ok=True)
                with open(path, "wb") as f:
                    while True:
                        chunk = r.read(65536)
                        if not chunk:
                            break
                        f.write(chunk)
                        done += len(chunk)
            except Exception:
                self._store.lib_del(th, lid)
                return False
            if done != total:
                self._store.lib_del(th, lid)
                return False
            self._fetch_failures = 0
            # Persisted size now exact.
            try:
                with self._store._lock:
                    rec = self._store._pairs.get(th)
                    if rec and lid in rec.get("lib", {}):
                        rec["lib"][lid]["bytes"] = done
                        self._store._save_locked()
            except Exception:
                pass
            return True

    def _count_failure(self):
        self._fetch_failures += 1
        if self._fetch_failures >= AV_CONSECUTIVE_BLOCKS:
            self._blocked_until = time.time() + AV_BACKOFF
            self._fetch_failures = 0


AV_HEADERS = {
    "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                   "AppleWebKit/537.36 (KHTML, like Gecko) "
                   "Chrome/124.0.0.0 Safari/537.36"),
    "Accept": "*/*",
}


class _Handler(BaseHTTPRequestHandler):
    server_version = "WavelineRelay/1.0"
    store: _Store = None
    pair_rate: _Rate = None
    sync_rate: _Rate = None
    cmd_rate: _Rate = None
    admin_rate: _Rate = None
    admin_key: str = ""  # set from RELAY_ADMIN_KEY; empty = admin disabled
    rooms: _Rooms = None
    lt_rate: _Rate = None
    join_rate: _Rate = None
    voice_rate: _Rate = None
    lib_rate: _Rate = None
    av_rate: _Rate = None       # AudioVault enqueue/status posts
    av_worker: "_AudioVaultWorker" = None
    # VPS agent (2026-10-01): server-side YouTube downloads, radio show
    # recordings and the per-account cookie store.  Created by serve()
    # and reset_store_for_tests(); handlers degrade honestly when absent
    # (old deployments that never built one).
    agent: "_vpsagent._VpsAgent" = None
    lr: "liveroom.LiveRooms" = None
    abuse: _Abuse = None
    lr_key: str = ""   # set from WAVELINE_LR_KEY; empty = Live Room disabled
    lr_rate: _Rate = None
    accounts: "_accounts.Accounts" = None   # None = accounts disabled
    mailer: "_accounts.Mailer" = None
    acct_rate: _Rate = None      # /account/api/* JSON endpoints
    page_rate: _Rate = None      # account/dashboard page GETs
    dash_rate: _Rate = None      # dashboard command posts
    # Security-alert mail pump (see _ALERTQ above): dies with the
    # process, and the events table re-alerts on the next start while
    # within its resend window.
    _alerts_pump_started = False
    _alerts_sweeper_started = False

    def __init__(self, *a, **kw):
        super().__init__(*a, **kw)
        if not _Handler._alerts_pump_started:
            _Handler._alerts_pump_started = True
            self._start_alert_pump()
        if not _Handler._alerts_sweeper_started:
            self.start_alert_sweeper()

    DIGEST_WEEK = 7 * 24 * 3600

    @classmethod
    def start_digest_sweeper(cls) -> None:
        """Hourly check: every account whose listen accumulator is a week
        old gets its digest mailed and the accumulator reset (entry 49).
        Never raises; mail failure just retries next hour."""
        def _digest_loop():
            while True:
                time.sleep(3600)
                try:
                    if cls.accounts is None or cls.mailer is None:
                        continue
                    for email in cls.store.digest_candidates(
                            cls.DIGEST_WEEK):
                        top = cls.store.listen_top(email)
                        total = cls.store.listen_total(email)
                        served = 0
                        j = cls.store.juke_get(email)
                        if j:
                            served = int(j.get("reqs_served", 0))
                        links = len(cls.accounts.linked_full(email))
                        if cls.mailer.send_digest(email, top, total,
                                                  served, links):
                            cls.store.listen_reset(email)
                except Exception:
                    pass
        threading.Thread(target=_digest_loop,
                         name="waveline-digest", daemon=True).start()

    @classmethod
    def start_alert_sweeper(cls) -> None:
        """One daemon thread sweeps recorded-but-unmailed security events
        (verify / reset / link / unlink - recorded inside Accounts) into
        the mail queue every DRAIN_INTERVAL seconds.  Idempotent."""
        if cls._alerts_sweeper_started:
            return
        cls._alerts_sweeper_started = True

        def _sweep() -> None:
            while True:
                time.sleep(DRAIN_INTERVAL)
                try:
                    if cls.accounts is None:
                        continue
                    for kind, email, detail in \
                            cls.accounts.drain_security_alerts():
                        _ALERTQ.append((kind, email, detail))
                        _ALERT_SIG.release()
                except Exception:
                    pass
        threading.Thread(target=_sweep, name="waveline-alert-sweep",
                         daemon=True).start()

    def _start_alert_pump(self) -> None:
        """One daemon thread drains the alert queue into SMTP, so a hung
        or slow mail server can NEVER stall a request thread (mail I/O
        timeouts live in the pump, not in the handler)."""
        def _pump() -> None:
            while True:
                _ALERT_SIG.acquire()
                try:
                    kind, email, detail = _ALERTQ.popleft()
                except IndexError:
                    continue
                try:
                    if _Handler.mailer is not None:
                        _Handler.mailer.send_alert(kind, email, detail)
                except Exception:
                    pass
        threading.Thread(target=_pump, name="waveline-alerts",
                         daemon=True).start()

    def queue_alert(self, kind: str, email: str, detail: str = "") -> None:
        """IMMEDIATE alert path: record the event pre-marked as mailed
        (the drain will never re-send it) and hand it to the background
        mail pump.  Used for events Accounts cannot see itself (pair).
        Never raises; the triggering action has already succeeded."""
        try:
            if self.accounts is not None:
                self.accounts.record_security_event(
                    kind, email, detail, sent=time.time())
            _ALERTQ.append((str(kind)[:40], str(email)[:190],
                            str(detail)[:160]))
            _ALERT_SIG.release()
        except Exception:
            pass

    # -- Live Room (host = operator key OR any paired Waveline user) --------
    LR_TOKEN_RE = re.compile(r"^[0-9a-f]{64}$")

    def _lr_auth(self):
        """Live Room host identity, or None.

        Two doors:
          * X-LR-Key == operator key          -> identity "admin"
          * X-LR-Token alone (account-only identity, 2026-09-27) ->
            identity "lr:<token-hash>" (any real Waveline user).
            The legacy X-LR-Code header is accepted but ignored.

        The room token derives from the identity, so one identity can
        only ever host ONE room (re-hosting refreshes it)."""
        if not self.lr:
            return None
        key = self.headers.get("X-LR-Key", "")
        if self.lr_key and key and hmac.compare_digest(key, self.lr_key):
            return "admin"
        token = self.headers.get("X-LR-Token", "")
        if TOKEN_RE.match(token):
            th = self._app_auth(token)
            if th:
                return "lr:" + th
        if self.abuse is not None:
            self.abuse.record("403", self._ip())   # failed LR host auth
        return None

    def _lr_room_token(self, identity: str) -> str:
        """Stable, unguessable 64-hex room token for a host identity.
        Not the identity itself: the room token is what the share link
        carries, and it must not reveal the pairing token hash."""
        return hashlib.sha256(
            ("waveline-lr-room\n" + identity).encode("utf-8")).hexdigest()

    def _raw_len(self) -> int:
        try:
            return int(self.headers.get("Content-Length", "0") or "0")
        except ValueError:
            return -1

    def _h_lr_create(self):
        identity = self._lr_auth()
        if identity is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad"}, 400)
        room_token = self._lr_room_token(identity)
        verdict = self.lr.create(room_token, str(data.get("title", ""))[:200],
                                 str(data.get("by", ""))[:24], identity)
        if verdict == "full":
            return self._json({"ok": False, "error": "busy"}, 503)
        self._json({"ok": True, "token": room_token})

    def _h_lr_push(self):
        identity = self._lr_auth()
        if identity is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        n = self._raw_len()
        if n <= 0 or n > LR_CHUNK_MAX:
            return self._json({"ok": False, "error": "bad"}, 400)
        try:
            chunk = self.rfile.read(n)
        except Exception:
            return self._json({"ok": False, "error": "bad"}, 400)
        verdict = self.lr.push(self._lr_room_token(identity), chunk)
        if verdict == "ok":
            return self._json({"ok": True})
        if verdict in ("missing", "full"):
            return self._json({"ok": False, "error": "closed"}, 404)
        return self._json({"ok": False, "error": "bad"}, 400)

    def _h_lr_stop(self):
        identity = self._lr_auth()
        if identity is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self.lr.close(self._lr_room_token(identity))
        self._json({"ok": True})

    def _h_lr_now(self):
        identity = self._lr_auth()
        if identity is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad"}, 400)
        self.lr.set_title(self._lr_room_token(identity),
                          str(data.get("title", ""))[:200])
        self._json({"ok": True})

    def _h_lr_stats(self):
        identity = self._lr_auth()
        if identity is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True, "stats": self.lr.stats(
            self._lr_room_token(identity))})

    def _h_lr_room_page(self, token: str):
        """Tiny dark, screen-reader-friendly listener page.  The audio
        itself comes from /lr/stream/<token> (audio/mpeg)."""
        body = (LR_PAGE % {"token": token, "title": "Waveline Live Room"})
        body = body.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _h_lr_meta(self, token: str):
        """Public read-only room info for the listener page poll."""
        m = self.lr.meta(token)
        if not m:
            return self._json({"ok": False, "error": "not found"}, 404)
        m["ok"] = True
        self._json(m)

    def _h_lr_stream(self, token: str):
        """Listener side: a plain chunked MP3 stream a browser <audio> tag
        (or any media player URL) can play.  Read-only: the token is a
        random 64-hex room token - it can never control or upload."""
        room = self.lr.open_stream(token)
        if room is None:
            return self._json({"ok": False, "error": "not found"}, 404)
        try:
            self.send_response(200)
            self.send_header("Content-Type", "audio/mpeg")
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            last_seen = -1
            idle = 0
            while idle < 60:                      # ~2 min of silence -> drop
                snap = room.snapshot_from(1)
                if snap is None:
                    time.sleep(1.0)
                    idle += 1
                    continue
                chunks, start = snap
                idle = 0
                take_from = last_seen - start + 1
                if take_from < 0:
                    take_from = 0
                for c in chunks[take_from:]:
                    self.wfile.write(c)
                if chunks:
                    last_seen = start + len(chunks) - 1
                if int(last_seen) % 8 == 0:       # cheap liveness probe
                    self.wfile.flush()
                time.sleep(0.25)
        except (BrokenPipeError, ConnectionResetError, OSError):
            pass
        finally:
            self.lr.release_stream(room)

    # -- admin (operator only; key never ships in the app or repo) ---------
    def _admin_ok(self) -> bool:
        if not self.admin_key:
            return False
        given = self.headers.get("X-Admin-Key", "")
        ok = hmac.compare_digest(given, self.admin_key)
        if not ok and self.abuse is not None:
            self.abuse.record("403", self._ip())
        return ok

    def _h_admin_list(self):
        if not self._admin_ok():
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True, "pairs": self.store.list_admin()})

    def _h_admin_abuse(self):
        """fail2ban-style daily summary: throttles, auth failures and
        forbidden hits per UTC day, with the top offending IPs."""
        if not self._admin_ok():
            return self._json({"ok": False, "error": "forbidden"}, 403)
        if self.abuse is None:
            return self._json({"ok": False, "error": "unavailable"}, 503)
        out = self.abuse.summary()
        out["ok"] = True
        self._json(out)

    def _h_admin_revoke(self):
        if not self._admin_ok():
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        code = str(data.get("code", ""))
        if code == "*":
            n = 0
            for row in self.store.list_admin():
                th = self.store.revoke_admin(row["th"])
                if th:
                    n += 1
                    try:
                        self.lr.close_by_owner("lr:" + th)
                    except Exception:
                        pass
            return self._json({"ok": True, "revoked": n})
        # ACCOUNT-ONLY identity: the argument is a session TOKEN HASH
        # (64 hex) from list_admin; the old 10-char code check is gone.
        if not self.TH_RE.match(code):
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self.store.revoke_admin(code)
        if th:
            try:
                self.lr.close_by_owner("lr:" + th)
            except Exception:
                pass
        self._json({"ok": True, "revoked": 1 if th else 0})

    _ADMIN_PAGE = r"""<!doctype html><html><head><meta charset="utf-8">
<title>Waveline relay admin</title></head><body>
<h1>Waveline relay admin</h1>
<p><label>Admin key <input type="password" id="k" size="40"></label>
<button id="go">List pairings</button>
<button id="abuse">Abuse summary</button></p>
<div id="out" aria-live="polite"></div>
<table id="t" border="1"><thead><tr>
<th>Client</th><th>Idle minutes</th><th>Now playing</th><th>Action</th>
</tr></thead><tbody id="rows"></tbody></table>
<div id="abusebox" style="display:none">
<h2>Abuse summary (UTC days, newest first)</h2>
<p id="abuseout" aria-live="polite"></p>
<table border="1"><thead><tr>
<th>Day</th><th>Rate limited (429)</th><th>Auth failures (404)</th>
<th>Forbidden (403)</th>
</tr></thead><tbody id="abuserows"></tbody></table>
<p style="font-weight:bold">Top offending IPs (most recent day)</p>
<table border="1"><thead><tr>
<th>IP</th><th>429</th><th>404</th><th>403</th>
</tr></thead><tbody id="iprows"></tbody></table>
<p>Auth failures are rejected code/token probes; forbidden are wrong
admin or Live Room keys. A handful per day is background noise; one IP
repeating every minute is someone probing you.</p>
</div>
<script>
var BASE=location.pathname.replace(/\/$/, '');
function h(){return {'X-Admin-Key':document.getElementById('k').value,
 'Content-Type':'application/json'}}
function list(){fetch(BASE+'/api/list',{headers:h()})
 .then(r=>r.json()).then(j=>{var o=document.getElementById('out');
 if(!j.ok){o.textContent='Wrong key or admin disabled';return}
 o.textContent=j.pairs.length+' active pairing(s)';
 var b=document.getElementById('rows');b.textContent='';
 j.pairs.forEach(function(p){var tr=document.createElement('tr');
 [(p.name||'(unnamed)'),p.idle_min,p.now_playing].forEach(function(v){
  var td=document.createElement('td');td.textContent=v;tr.appendChild(td)});
 var td=document.createElement('td');var btn=document.createElement('button');
 btn.textContent='Revoke '+(p.name||p.th.slice(0,8));
 btn.onclick=function(){revoke(p.th)};td.appendChild(btn);tr.appendChild(td);
 b.appendChild(tr)})})
 .catch(function(){document.getElementById('out').textContent=
  'Network error - could not reach the relay API'});}
function revoke(code){fetch(BASE+'/api/revoke',{method:'POST',headers:h(),
 body:JSON.stringify({code:code})}).then(()=>list())}
function abuse(){fetch(BASE+'/api/abuse',{headers:h()})
 .then(r=>r.json()).then(j=>{var o=document.getElementById('abuseout');
 var box=document.getElementById('abusebox');
 if(!j.ok){o.textContent='Wrong key or admin disabled';box.style.display='block';return}
 var rows=j.days||[];
 o.textContent=rows.length?
  ('Tracked IPs today: '+(j.tracked_ips||0)):
  'No abuse recorded yet - quiet relay.';
 var b=document.getElementById('abuserows');b.textContent='';
 rows.forEach(function(d){var tr=document.createElement('tr');
 [d.date,d.rate_limited,d.auth_failures,d.forbidden].forEach(function(v){
  var td=document.createElement('td');td.textContent=v;tr.appendChild(td)});
 b.appendChild(tr)});
 var ib=document.getElementById('iprows');ib.textContent='';
 var top=(rows[0]&&rows[0].top_ips)||[];
 top.forEach(function(t){var tr=document.createElement('tr');
 [t.ip,t.rate_limited,t.auth_failures,t.forbidden].forEach(function(v){
  var td=document.createElement('td');td.textContent=v;tr.appendChild(td)});
 ib.appendChild(tr)});
 box.style.display='block'})
 .catch(function(){var o=document.getElementById('abuseout');
  o.textContent='Network error - could not reach the relay API';
  document.getElementById('abusebox').style.display='block'});}
document.getElementById('go').onclick=list;
document.getElementById('abuse').onclick=abuse;
</script></body></html>"""

    def _h_admin_page(self):
        if not self.admin_key:
            return self._reject()
        body = self._ADMIN_PAGE.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *a):  # silence default stderr chatter
        pass

    # -- helpers ----------------------------------------------------------
    def _ip(self) -> str:
        # SECURITY (2026-09-23): the FIRST X-Forwarded-For entry is fully
        # attacker-controlled (a direct client can send any header, and
        # Apache APPENDS the real client IP to what the client sent), so
        # keying rate buckets on it let anyone rotate forged values and
        # evade every limiter (fresh bucket per request).  Trust XFF only
        # for loopback peers - i.e. our own reverse proxy, which appends
        # the real client IP as the LAST entry - and only if it parses as
        # an IP.  Everyone else is rate-limited by their TCP peer address,
        # which cannot be forged.  Same policy as relay_server.py.
        return _effective_ip(self.client_address[0],
                             self.headers.get("X-Forwarded-For", ""))

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

    def _reject(self):
        # IDENTICAL body for unknown/unpaired/expired: no code-guess oracle.
        if self.abuse is not None:
            self.abuse.record("404", self._ip())
        self._json({"ok": False, "error": "not found"}, 404)

    def _body(self):
        data = self._body_with_limit(MAX_BODY)
        # "too large" is an internal signal for the agent routes;
        # every other handler keeps the exact old behavior (None).
        return None if data == "too large" else data

    def _content_length(self) -> int:
        try:
            return int(self.headers.get("Content-Length", "0") or "0")
        except ValueError:
            return -1

    def _body_with_limit(self, cap: int):
        """Parse a JSON dict body of at most `cap` bytes.

        A declared Content-Length over the cap returns the string "too
        large" (not None) so callers can answer honestly instead of
        mistaking a big body for a bad one.  Bodies WITHIN the cap that
        fail to parse stay None ("bad request"), as before.
        """
        n = self._content_length()
        if n <= 0:
            return None
        if n > cap:
            return "too large"
        try:
            data = json.loads(self.rfile.read(n).decode("utf-8"))
        except Exception:
            return None
        if not isinstance(data, dict):
            return None
        return data

    def _throttled(self):
        """429 helper: records the hit in the abuse tracker, then answers
        with a body that says only "slow down" (no leak)."""
        if self.abuse is not None:
            self.abuse.record("429", self._ip())
        return self._json({"ok": False, "error": "slow down"}, 429)

    def _h_pair(self):
        """ACCOUNT-ONLY identity (owner order 2026-09-27): the app's
        DPAPI-protected token is the whole credential - no pairing codes
        are generated, announced, rotated or typed anywhere.  A code in
        the payload is ignored (older clients, tests)."""
        if not self.pair_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        token = str(data.get("token", ""))
        name = str(data.get("name", ""))[:40]
        if not TOKEN_RE.match(token):
            return self._json({"ok": False, "error": "bad request"}, 400)
        self.store.pair(token, name=name)
        self._json({"ok": True})

    def _h_sync(self):
        if not self.sync_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        token = str(data.get("token", ""))
        th = self._app_auth(token)
        if th is None:
            return self._reject()
        raw_state = data.get("state")
        state = {}
        if isinstance(raw_state, dict):
            for k in STATE_KEYS & set(raw_state):
                v = raw_state[k]
                if k == "now_playing":
                    state[k] = str(v)[:MAX_STR]
                elif k == "playing":
                    state[k] = bool(v)
                elif k == "live":
                    state[k] = bool(v)
                elif k == "url":
                    state[k] = str(v)[:MAX_URL]
                elif k in ("volume", "position"):
                    try:
                        state[k] = max(0.0, float(v))
                    except (TypeError, ValueError):
                        pass
                elif k == "alarm_active":
                    state[k] = bool(v)
                elif k == "alarm_label":
                    state[k] = str(v)[:MAX_STR]
        cmds, devices = self.store.sync(th, state)
        # In-app notifications (5.1): deliver anything queued for the
        # linked account on this poll (owner feedback alerts etc.).
        # NOTE: no VALID_ACTIONS filter here - entry points already
        # validate (phones via _h_cmd's whitelist).  Delivery must be
        # faithful so app-only actions reach the app.
        # Jukebox now-playing mirror (entry 49): if this account has a
        # live share, guests see what is playing (name only - no paths).
        email = self._acct_email_for_th(th)
        if email and self.store.juke_get(email):
            self.store.juke_set_now(email, {
                "name": state.get("now_playing", ""),
                "playing": "yes" if state.get("playing") else "no"})
        notifs = []
        try:
            email = self._acct_email_for_th(th)
            if email:
                notifs = self.accounts.drain_notifications(email)
        except Exception:
            notifs = []
        self._json({"ok": True, "commands": cmds, "devices": devices,
                    "notifications": notifs})

    def _h_device_revoke(self):
        """Owner revokes one (or all) control devices for their pairing."""
        if not self.sync_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        n = self.store.ban_device_by_th(th, str(data.get("device", "")))
        self._json({"ok": True, "revoked": n})

    def _h_cmd(self, code):
        if not self._pc_ref_ok(code):
            return self._reject()
        if not self.cmd_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        action = str(data.get("action", ""))
        if action not in VALID_ACTIONS:
            return self._json({"ok": False, "error": "bad request"}, 400)
        # Account session (phone page under /account/remote/<th>): push
        # straight onto the linked PC - no device headers involved.
        th = self._ensure_account_pc(code)
        if th:
            result = self.store.push_by_th(th, action)
            return self._json({"ok": result == "ok"})
        devid = str(data.get("device", "") or
                    self.headers.get("X-Device-Id", ""))[:64]
        name = str(data.get("name", "") or
                   self.headers.get("X-Device-Name", ""))
        if devid:
            seen = self.store.device_seen(code, devid, name)
            if seen == "banned":
                return self._json(
                    {"ok": False, "error": "device-revoked"}, 403)
            if seen == "used":
                return self._json(
                    {"ok": False, "error": "pairing-closed"}, 403)
            if seen == "ok":
                # A NEW control device just registered (device_seen returns
                # 'ok' for new and repeat connections alike; the store knows
                # which - this call is a no-op for known devices).
                self.queue_pair_alert_if_new(code, devid, name)
        result = self.store.push(code, action, devid=devid)
        if result == "ok":
            self._json({"ok": True})
        elif result == "banned":
            return self._json({"ok": False, "error": "device-revoked"}, 403)
        elif result == "closed":
            return self._json({"ok": False, "error": "pairing-closed"}, 403)
        else:
            self._reject()

    def _h_state(self, code):
        if not self._pc_ref_ok(code):
            return self._reject()
        if not self.cmd_rate.allow(self._ip()):
            return self._throttled()
        # Account session (phone page under /account/remote/<th>).
        th = self._ensure_account_pc(code)
        if th:
            return self._json({"ok": True, "state": self.store.state_by_th(th)})
        devid = str(self.headers.get("X-Device-Id", ""))[:64]
        if devid:
            seen = self.store.device_seen(
                code, devid, self.headers.get("X-Device-Name", ""))
            if seen == "banned":
                return self._json(
                    {"ok": False, "error": "device-revoked"}, 403)
        st = self.store.state(code)
        if not st:
            # Distinguish "code never paired / expired" (404) from
            # "paired, but the owner's PC has not synced yet" (200 with
            # an empty state) - now keyed by TH or legacy code.
            if not self.store.paired(code):
                return self._reject()
            st = {}
        self._json({"ok": True, "state": st})

    def _h_revoke(self):
        """App-side goodbye.  ACCOUNT-ONLY identity: the token alone (the
        code field of older clients is ignored)."""
        if not self.cmd_rate.allow(self._ip()):
            # 429 only says "slow down" - it cannot leak code validity,
            # so the always-ok body below stays intact for allowed calls.
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        token = str(data.get("token", ""))
        if TOKEN_RE.match(token):
            th = self.store.revoke_token(token)
            if th:
                # A revoked user's Live Room must die with the session.
                try:
                    self.lr.close_by_owner("lr:" + th)
                except Exception:
                    pass
        # always ok: revocation must not leak whether a code existed
        self._json({"ok": True})

    # -- routing -----------------------------------------------------------
    # Account pages: dark, semantic, screen-reader first (labels, real
    # headings, aria-live status) - the owner navigates them with NVDA.
    _ACCT_STYLE = """<style>
body{font-family:system-ui,sans-serif;background:#0f1319;color:#e8eaed;
 max-width:34rem;margin:6vh auto;padding:0 1.2rem}
h1{font-size:1.6em}h2{font-size:1.25em;margin-top:1.4em}
.box{background:#171d26;border-radius:14px;padding:1.1rem 1rem;margin:1rem 0}
label{display:block;margin:.7rem 0 .25rem}
input{font-size:1.05em;width:100%;box-sizing:border-box;background:#0f1319;
 color:#e8eaed;border:2px solid #2a3644;border-radius:10px;padding:.45em}
button,button.main{font-size:1.05em;background:#4a90d9;color:#fff;border:0;
 border-radius:10px;padding:.55em 1.2em;margin-top:.8rem}
button.secondary{background:#1f2937}
#msg{min-height:1.3em;font-weight:600}
.ok{color:#7ddc8f}.err{color:#ff8a80}
.hint{color:#9aa4b0;font-size:.92em}
table{border-collapse:collapse;width:100%}
th,td{text-align:left;padding:.35rem .5rem;border-bottom:1px solid #232c38}
.on{color:#7ddc8f}.off{color:#9aa4b0}
</style>"""

    _ACCT_PAGE = """<!doctype html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>%(title)s</title>%(style)s</head><body>
<h1>%(h1)s</h1>%(body)s
<p class="hint">Waveline remote access</p>
</body></html>"""

    def _h_acct_redirect(self, target: str) -> None:
        """Canonicalise dashboard URLs so every relative link in the
        pages resolves under /account/.  'account/dashboard' is correct
        from BOTH /remote/account (ingress) and /account (tests): a
        relative Location resolves against the directory of the current
        path, which is <ingress>/ in both cases."""
        self.send_response(301)
        self.send_header("Location", "account/" + target)
        self.send_header("Content-Length", "0")
        self.end_headers()

    def _acct_page(self, title: str, h1: str, body: str) -> None:
        html = self._ACCT_PAGE % {"title": title, "style": self._ACCT_STYLE,
                                  "h1": h1, "body": body}
        data = html.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    @staticmethod
    def _msg(ok: bool, text: str) -> str:
        cls = "ok" if ok else "err"
        return '<p id="msg" class="%s" role="status" aria-live="polite">%s</p>' % (
            cls, text)

    @staticmethod
    def _field(name: str, label: str, typ: str = "text",
               autocomplete: str = "off") -> str:
        return ('<label for="%s">%s</label>'
                '<input id="%s" name="%s" type="%s" autocomplete="%s">'
                % (name, label, name, name, typ, autocomplete))

    # -- cookies (OWASP Session Management) --------------------------------
    SESSION_COOKIE = "wlsession"

    def _session_from_cookie(self) -> str:
        raw = self.headers.get("Cookie", "")
        for part in raw.split(";"):
            k, _, v = part.strip().partition("=")
            if k == self.SESSION_COOKIE:
                return v.strip()
        return ""

    def _cookie_secure(self) -> bool:
        """OWASP: Secure is mandatory over TLS.  Production ingress is
        Apache HTTPS; direct plain-HTTP access (tests, local dev) must
        still work, so: explicit env wins, else trust X-Forwarded-Proto,
        else default to Secure (production posture)."""
        v = os.environ.get("WAVELINE_ACCOUNTS_COOKIE_SECURE",
                           "").strip().lower()
        if v in ("0", "false", "no"):
            return False
        if v in ("1", "true", "yes"):
            return True
        proto = self.headers.get("X-Forwarded-Proto", "").strip().lower()
        if proto:
            return proto == "https"
        return True

    def _set_session_cookie(self, token: str, remember: bool = False) -> None:
        """OWASP cookie hygiene.  remember=False -> NO Max-Age: the
        browser drops it when the browser closes (a "session cookie",
        server-side 30-day inactivity backstop applies).  remember=True
        -> persistent: Max-Age tracks the server-side sliding year."""
        if remember:
            max_age = "; Max-Age=%d" % _accounts.PERSISTENT_TTL
        else:
            max_age = ""
        self.send_header(
            "Set-Cookie",
            "%s=%s; Path=/%s; HttpOnly; %sSameSite=Lax"
            % (self.SESSION_COOKIE, token, max_age,
               "Secure; " if self._cookie_secure() else ""))

    def _clear_session_cookie(self) -> None:
        self.send_header(
            "Set-Cookie",
            "%s=; Path=/; Max-Age=0; HttpOnly; %sSameSite=Lax"
            % (self.SESSION_COOKIE,
               "Secure; " if self._cookie_secure() else ""))

    def _acct_user(self):
        """Session -> user dict or None."""
        tok = self._session_from_cookie()
        if not tok:
            return None
        return self.accounts.auth_session(tok)

    # -- app identity helpers (token-authed, like /app/sync) ----------------
    def _app_auth(self, token: str):
        """Pairing token -> token hash, or None.  The app proves ownership
        of its pairing exactly like /app/sync; the raw token is never
        stored."""
        if not TOKEN_RE.match(str(token or "")):
            return None
        return self.store.auth_token(token)

    def _acct_email_for_th(self, th: str) -> str:
        """Account email a token hash is linked to ('' = none)."""
        for email, th_owned in self.accounts.linked_full_emails():
            if hmac.compare_digest(th_owned, str(th or "")):
                return email
        return ""

    # -- JSON endpoints (uniform bodies: no enumeration oracles) ----------
    _GENERIC_OK = {"ok": True, "message":
                   "If that address can have an account, the next step "
                   "is in your inbox."}
    _GENERIC_FAIL = {"ok": False, "error": "bad request"}

    def _h_account_register(self):
        data = self._body()
        if not data:
            return self._json(self._GENERIC_FAIL, 400)
        email = str(data.get("email", ""))
        nick = str(data.get("nickname", ""))
        password = str(data.get("password", ""))
        if len(password) > _accounts.PASSWORD_MAX:
            return self._json(self._GENERIC_FAIL, 400)
        token, problem = self.accounts.register(email, nick, password)
        if token:
            if not self.mailer.send_verify(email, token):
                # journald-visible; the token can be re-sent later
                self.log_message("verification mail to %s failed", email)
        if problem in ("too-short", "too-long", "too-common"):
            # Feedback about the typed input only - never about accounts.
            hints = {"too-short": "Use at least %d characters."
                     % _accounts.PASSWORD_MIN,
                     "too-long": "That password is too long.",
                     "too-common": "That password is too common - "
                                   "choose something unique."}
            return self._json({"ok": False, "error": hints[problem]}, 400)
        return self._json(self._GENERIC_OK)

    def _h_account_login(self):
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        tok = self.accounts.login(str(data.get("email", "")),
                                  str(data.get("password", "")),
                                  remember=bool(data.get("remember")))
        if not tok:
            return self._json({"ok": False,
                               "error": "Login failed; invalid email or "
                                        "password."}, 403)
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self._set_session_cookie(tok, remember=bool(data.get("remember")))
        body = json.dumps({"ok": True}).encode("utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _h_account_logout(self):
        tok = self._session_from_cookie()
        if tok:
            self.accounts.logout(tok)
        self._clear_session_cookie()
        self._json({"ok": True})

    def _h_account_forgot(self):
        data = self._body()
        if not data:
            return self._json(self._GENERIC_FAIL, 400)
        email = str(data.get("email", ""))
        token = self.accounts.request_reset(email)
        if token:
            self.mailer.send_reset(email.strip().lower(), token)
        return self._json(self._GENERIC_OK)

    def _h_account_reset(self):
        data = self._body()
        if not data:
            return self._json(self._GENERIC_FAIL, 400)
        problem = self.accounts.reset(str(data.get("token", "")),
                                      str(data.get("password", "")))
        if problem:
            msgs = {"bad-token": "This reset link is invalid or expired.",
                    "too-short": "Use at least %d characters."
                                 % _accounts.PASSWORD_MIN,
                    "too-long": "That password is too long.",
                    "too-common": "That password is too common."}
            return self._json({"ok": False,
                               "error": msgs.get(problem, "bad request")}, 400)
        try:
            self.mailer.send_reset_done(str(data.get("email", "")))
        except Exception:
            pass
        return self._json({"ok": True})

    def _h_account_link(self):
        """App <-> account device link.  Authed by the APP's pairing
        token (the app proves ownership of its pairing, exactly like
        /app/sync); the short code says WHICH account.  Never stores the
        raw token - only the relay's standard hash."""
        if not self.sync_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        code = str(data.get("code", ""))
        token = str(data.get("token", ""))
        if not TOKEN_RE.match(token):
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self.store.auth_token(token)   # token-only identity (rotated
        if th is None:                      # invitation codes stay valid)
            return self._json({"ok": False, "error": "bad request"}, 400)
        nick = self.accounts.consume_link_code(
            code, th, pc_name=str(data.get("pc_name", "")))
        if nick is None:
            # same answer whether the code is wrong, expired or used
            return self._json({"ok": False,
                               "error": "Link code invalid or expired - "
                                        "generate a new one in the "
                                        "dashboard."}, 400)
        # link is recorded inside consume_link_code; the alert sweep
        # emails it (one mailing path for account lifecycle events)
        self._json({"ok": True, "account": nick})

    def _h_account_unlink(self):
        if not self.sync_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        code = str(data.get("code", ""))
        token = str(data.get("token", ""))
        if not TOKEN_RE.match(token):
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self.store.auth_token(token)
        if th is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        n = self.accounts.unlink_by_hash(th)
        # unlink is recorded inside unlink_by_hash; alert sweep emails it
        self._json({"ok": True, "unlinked": n})

    # -- settings sync (per-account blob; app-authed) ------------------------
    def _h_settings_get(self):
        """App pulls the account's settings blob.  Always 200 (errors in
        the body) so the client treats no-account like an empty blob."""
        if not self.sync_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        email = self._acct_email_for_th(th)
        blob = self.accounts.settings_get(email) if email else None
        if blob is None:
            return self._json({"ok": True, "settings": None,
                               "no_account": True})
        return self._json({"ok": True, "settings": blob,
                           "no_account": False})

    def _h_settings_put(self):
        """App pushes its whole settings blob (selected keys, cleaned
        client-side).  Never blocking, never partial: the blob replaces
        the stored one atomically."""
        if not self.sync_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        email = self._acct_email_for_th(th)
        if not email:
            return self._json({"ok": False, "error": "no account linked"},
                              404)
        blob = data.get("settings")
        if not isinstance(blob, dict):
            return self._json({"ok": False, "error": "bad request"}, 400)
        if len(json.dumps(blob).encode("utf-8")) > MAX_BLOB:
            return self._json({"ok": False, "error": "too large"}, 413)
        if self.accounts.settings_put(email, blob):
            return self._json({"ok": True})
        return self._json({"ok": False, "error": "rejected"}, 400)

    # -- pages --------------------------------------------------------------
    # -- user feedback (5.1) -------------------------------------------------
    FEEDBACK_KINDS = ("suggestion", "bug", "question", "other")

    def _h_feedback_page(self):
        options = "".join(
            '<option value="%s">%s</option>' % (k, k.title())
            for k in self.FEEDBACK_KINDS)
        body = (
            "<p>Something you love? Something broken? Tell us - every "
            "message reaches the Waveline developer by email. Add an "
            "email address if you would like a reply.</p>"
            '<form method="post" action="feedback">'
            '<label for="fkind">Topic</label>'
            '<select id="fkind" name="kind">%s</select>'
            % options
            + self._field("fcontact", "Your email (optional, for a reply)",
                          "email", "email")
            + '<label for="fmessage">Message</label>'
              '<textarea id="fmessage" name="message" rows="8" '
              'maxlength="20000" required></textarea>'
              '<p><button type="submit">Send feedback</button></p></form>')
        self._acct_page("Waveline feedback", "Send feedback", body)

    def _h_feedback_submit(self):
        """Accept the feedback from the site form (urlencoded) or the app
        (JSON, optionally with a redacted log snapshot).  Mails both
        operator addresses; answers the app in JSON and the site with a
        page.  Delivery failures are honest, never silent lies."""
        cap = 400_000
        ln = int(self.headers.get("Content-Length", "0") or "0")
        if ln <= 0 or ln > cap:
            return self._json({"ok": False, "error": "invalid size"},
                              code=400)
        raw = self.rfile.read(ln).decode("utf-8", "replace")
        ctype = (self.headers.get("Content-Type") or
                 "").split(";")[0].strip().lower()
        wants_json = ctype == "application/json" or self.path.endswith("/api")
        if ctype == "application/json":
            import json as _json
            try:
                data = _json.loads(raw or "{}")
            except Exception:
                return self._json({"ok": False, "error": "bad json"},
                                  code=400)
            if not isinstance(data, dict):
                return self._json({"ok": False, "error": "bad json"},
                                  code=400)
        else:
            from urllib.parse import parse_qs as _pq
            data = {k: v[0] for k, v in _pq(raw).items()}
        kind = str(data.get("kind", "other"))[:20]
        if kind not in self.FEEDBACK_KINDS:
            kind = "other"
        contact = str(data.get("contact", ""))[:200].strip()
        message = str(data.get("message", "")).strip()
        version = str(data.get("version", ""))[:40].strip()
        log_snapshot = str(data.get("log", ""))
        if len(message) < 3:
            problem = "Please include a short message."
            if wants_json:
                return self._json({"ok": False, "error": problem}, code=400)
            return self._acct_page(
                "Waveline feedback", "Check the form",
                self._msg(False, problem) + self._h_feedback_form_only())
        if len(message) > 20000:
            message = message[:20000]
        if contact and not _accounts._EMAIL_ADDR_RE.match(contact):
            problem = ("That email address does not look valid - fix it "
                       "or leave it empty.")
            if wants_json:
                return self._json({"ok": False, "error": problem}, code=400)
            return self._acct_page(
                "Waveline feedback", "Check the form",
                self._msg(False, problem) + self._h_feedback_form_only())
        ticket = _accounts.Mailer.new_ticket_id()
        # msg_id threads the ack (and any later operator reply) under
        # the notification copy that lands in Gmail/webmail.
        msg_id = self.mailer.send_feedback(
            ticket, kind, contact, message, version, log_snapshot)
        if not msg_id:
            self.log_message("feedback %s: mail delivery FAILED", ticket)
        elif contact and _accounts._EMAIL_ADDR_RE.match(contact):
            try:
                self.mailer.send_feedback_ack(ticket, contact,
                                              in_reply_to=msg_id)
            except Exception:
                pass
        # Persist for the in-app feedback browser (5.1): helpers and
        # admins read reports without server access.
        try:
            self.accounts.feedback_add(ticket, kind, contact, message,
                                       version, log_snapshot)
        except Exception:
            pass
        # In-app notification for the owner (5.1): visible the next
        # time Waveline opens (the app drains notifications on sync).
        try:
            self.accounts.notify_role(
                "owner", "feedback",
                "New feedback: %s" % kind,
                (message[:200] + ("..." if len(message) > 200 else "")),
                ticket=ticket)
        except Exception:
            pass
        if wants_json:
            return self._json({"ok": bool(msg_id), "ticket": ticket})
        if msg_id:
            status = ("Thank you! Your feedback id is %s - the developer "
                      "reads every message." % ticket)
        else:
            status = ("Sorry - sending failed just now. Please try again "
                      "in a little while.")
        return self._acct_page("Waveline feedback",
                               "Feedback sent" if msg_id
                               else "Sending failed",
                               self._msg(bool(msg_id), status))

    def _h_feedback_form_only(self) -> str:
        options = "".join(
            '<option value="%s">%s</option>' % (k, k.title())
            for k in self.FEEDBACK_KINDS)
        return (
            '<form method="post" action="feedback">'
            '<label for="fkind">Topic</label>'
            '<select id="fkind" name="kind">%s</select>' % options
            + self._field("fcontact", "Your email (optional, for a reply)",
                          "email", "email")
            + '<label for="fmessage">Message</label>'
              '<textarea id="fmessage" name="message" rows="8" '
              'maxlength="20000" required></textarea>'
              '<p><button type="submit">Send feedback</button></p></form>')

    def _h_acct_signup(self):
        body = ("""<p>Create your Waveline account. We will email you a
 confirmation link - clicking it is what makes the account real.</p>"""
                + self._h_acct_signup_form_only()
                + '<p class="hint">Already have one? '
                  '<a href="login" style="color:#4a90d9">Log in</a></p>')
        self._acct_page("Waveline account", "Create your account", body)

    def _h_acct_login_page(self):
        body = (self._h_acct_login_form_only()
                + '<p class="hint">Forgot it? '
                  '<a href="forgot" style="color:#4a90d9">Reset your '
                  'password</a> &middot; No account? '
                  '<a href="signup" style="color:#4a90d9">Sign up</a></p>')
        self._acct_page("Waveline account", "Log in", body)

    def _h_acct_forgot_page(self):
        body = ("""<p>Enter your email and we will send a reset link
 (whether or not an account exists - the answer looks the same).</p>
<form method="post" action="forgot">"""
                + self._field("email", "Email", "email", "email")
                + '<p><button type="submit">Send reset link</button></p></form>')
        self._acct_page("Waveline account", "Reset your password", body)

    def _handle_account_form(self, kind: str) -> None:
        """HTML form posts reuse the same logic as the JSON endpoints and
        answer with a page + aria-live status (NVDA announces it)."""
        ln = int(self.headers.get("Content-Length", "0") or "0")
        if ln <= 0 or ln > _accounts.PASSWORD_MAX * 4:
            return self._acct_page("Waveline account", "Oops",
                                   self._msg(False, "Invalid submission."))
        from urllib.parse import parse_qs as _pq
        form = {k: v[0] for k, v in
                _pq(self.rfile.read(ln).decode("utf-8", "replace")).items()}
        if kind == "signup":
            # Confirm-password (user request, standard everywhere):
            # mismatch is feedback about the TYPED INPUT only.
            if form.get("password", "") != form.get("password2", ""):
                return self._acct_page(
                    "Waveline account", "Check the form",
                    self._msg(False, "The two passwords do not match."))
            token, problem = self.accounts.register(
                form.get("email", ""), form.get("nickname", ""),
                form.get("password", ""))
            if token:
                if not self.mailer.send_verify(form.get("email", ""), token):
                    self.log_message("verification mail to %s failed",
                                     form.get("email", ""))
            if problem in ("too-short", "too-long", "too-common"):
                hints = {"too-short": "Use at least %d characters."
                         % _accounts.PASSWORD_MIN,
                         "too-long": "That password is too long.",
                         "too-common": "That password is too common - "
                                       "choose something unique."}
                return self._acct_page(
                    "Waveline account", "Check the form",
                    self._msg(False, hints[problem])
                    + self._h_acct_signup_form_only())
            return self._acct_page(
                "Waveline account", "Check your email",
                self._msg(True, "If that address can have an account, the "
                                "next step is in your inbox."))
        if kind == "login":
            tok = self.accounts.login(form.get("email", ""),
                                      form.get("password", ""),
                                      remember=bool(form.get("remember")))
            if not tok:
                return self._acct_page(
                    "Waveline account", "Log in",
                    self._msg(False, "Login failed; invalid email or "
                                  "password.")
                    + self._h_acct_login_form_only(
                        form.get("email", ""),
                        bool(form.get("remember"))))
            self.send_response(303)
            self._set_session_cookie(tok,
                                     remember=bool(form.get("remember")))
            self.send_header("Location", "dashboard")
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        if kind == "forgot":
            token = self.accounts.request_reset(form.get("email", ""))
            if token:
                self.mailer.send_reset(form.get("email", "").strip().lower(),
                                       token)
            return self._acct_page(
                "Waveline account", "Check your email",
                self._msg(True, "If that address can have an account, a "
                                "reset link is in your inbox."))
        if kind == "reset":
            # Confirm-password first, and a mismatch does NOT consume the
            # single-use token (the user just fixes the typo and retries).
            if form.get("password", "") != form.get("password2", ""):
                body = (self._msg(False, "The two passwords do not match.")
                        + '<form method="post" action="reset">'
                        + '<input type="hidden" name="token" value="'
                        + str(form.get("token", "")).replace(
                            '&', '&amp;').replace('"', '&quot;') + '">'
                        + self._field("password", "New password", "password",
                                      "new-password")
                        + self._field("password2", "Confirm new password",
                                      "password", "new-password")
                        + '<p><button type="submit">Set new password'
                          '</button></p></form>')
                return self._acct_page(
                    "Waveline account", "Set a new password", body)
            problem = self.accounts.reset(form.get("token", ""),
                                          form.get("password", ""))
            if problem:
                msgs = {"bad-token": "This reset link is invalid or expired.",
                        "too-short": "Use at least %d characters."
                                     % _accounts.PASSWORD_MIN,
                        "too-long": "That password is too long.",
                        "too-common": "That password is too common."}
                return self._acct_page(
                    "Waveline account", "Set a new password",
                    self._msg(False, msgs.get(problem, "Invalid input.")))
            return self._acct_page(
                "Waveline account", "Password changed",
                self._msg(True, "Password changed. All devices were signed "
                                "out - log in again with the new password."))

    def _h_acct_signup_form_only(self):
        return ("<form method=\"post\" action=\"signup\">"
                + self._field("email", "Your email", "email", "email")
                + self._field("nickname", "Nickname (what devices call you)")
                + self._field("password", "Choose a password (10+ "
                              "characters)", "password", "new-password")
                + self._field("password2", "Confirm password", "password",
                              "new-password")
                + '<p><button type="submit">Create account</button></p></form>')

    def _h_acct_login_form_only(self, email: str = "",
                                remember: bool = False):
        checked = " checked" if remember else ""
        return ('<form method="post" action="login">'
                + self._field("email", "Email", "email", "email")
                + self._field("password", "Password", "password",
                              "current-password")
                + '<p><label><input type="checkbox" name="remember" '
                  'value="1"%s> Remember me on this browser</label></p>' % checked
                + '<p><button type="submit">Log in</button></p></form>')

    def _h_acct_verify(self, query: str):
        qs = parse_qs(query)
        token = (qs.get("token", [""])[0])[:200]
        ok = self.accounts.verify(token) if token else False
        if ok:
            body = (self._msg(True, "Email confirmed. Your account is "
                                    "active - log in on any device.")
                    + '<p><a href="login" style="color:#4a90d9">'
                      'Log in</a></p>')
        else:
            body = (self._msg(False, "This confirmation link is invalid or "
                                    "expired.")
                    + '<p class="hint">Ask for a new email on the '
                      '<a href="login" style="color:#4a90d9">log in</a> '
                      'page.</p>')
        self._acct_page("Waveline account", "Email confirmation", body)

    def _h_acct_reset_page(self, query: str):
        qs = parse_qs(query)
        token = (qs.get("token", [""])[0])[:200]
        body = ('<form method="post" action="reset">'
                + '<input type="hidden" name="token" value="'
                + token.replace('&', '&amp;').replace('"', '&quot;') + '">'
                + self._field("password", "New password (10+ characters)",
                              "password", "new-password")
                + self._field("password2", "Confirm new password",
                              "password", "new-password")
                + '<p><button type="submit">Set new password</button></p>'
                + '</form>')
        self._acct_page("Waveline account", "Set a new password", body)

    def _h_acct_dashboard(self):
        user = self._acct_user()
        if user is None:
            return self._acct_page(
                "Waveline account", "Please log in",
                self._msg(False, "Your session expired - log in again.")
                + self._h_acct_login_form_only())
    # Full command set (mirrors the phone page; all whitelisted)
    _DASH_CMDS = (("play_pause", "Play / pause"), ("stop", "Stop"),
                  ("next_track", "Next track"), ("prev_track", "Previous"),
                  ("seek_back", "Seek back"), ("seek_forward", "Seek forward"),
                  ("volume_up", "Volume +"), ("volume_down", "Volume \u2212"),
                  ("mute", "Mute"), ("toggle_shuffle", "Shuffle"),
                  ("cycle_repeat", "Repeat"), ("bookmark_add", "Bookmark"),
                  ("sleep_timer", "Sleep timer"), ("announce_time", "Say time"))

    def _h_acct_dashboard(self):
        user = self._acct_user()
        if user is None:
            return self._acct_page(
                "Waveline account", "Please log in",
                self._msg(False, "Your session expired - log in again.")
                + self._h_acct_login_form_only())
        rows = []
        for th, label in self._linked_pcs(user["email"]):
            st = self.store.state_by_th(th)
            devs = self.store.devices_by_th(th)
            online = self.store.pc_online(th)
            np = st.get("now_playing", "(nothing yet - open Waveline)")
            playing = "playing" if st.get("playing") else "paused"
            cmds = "".join(
                '<button data-cmd="%s" data-pc="%s">%s</button>'
                % (a, th, lbl) for a, lbl in self._DASH_CMDS)
            devrows = "".join(
                '<tr><td>%s</td><td class="%s">%s (idle %d min)</td>'
                '<td><button class="secondary" data-ren="%s" '
                'data-pc="%s">Rename</button> '
                '<button class="rec" data-ban="%s" data-pc="%s">Unpair'
                '</button></td></tr>'
                % (d["name"], "on" if d["online"] else "off",
                   "online" if d["online"] else "offline",
                   d["idle_min"], d["id"], th, d["id"], th)
                for d in devs) or (
                '<tr><td colspan="3">No control devices '
                'paired yet - pair a phone with a relay '
                'code, then it appears here.</td></tr>')
            rows.append(
                '<section class="pc" data-pc="%s">'
                '<h2><span class="pcname">%s</span> '
                '<span class="%s">%s</span> '
                '<button class="secondary pcren" data-pcren="%s">Rename '
                'PC</button></h2>'
                '<div class="box"><p class="np">Now playing: %s (%s)</p>'
                '<p>%s</p></div>'
                '<h3>Stored music (cloud library)</h3>'
                '<div class="lib" data-pc="%s"></div>'
                '<p><button class="secondary" data-cont="%s">Continue '
                'on this device</button> <span class="hint" '
                'data-contst="%s"></span></p>'
                % (th, label, "on" if online else "off",
                   "online" if online else "offline", th, np, playing, cmds,
                   th, th, th)
                + '<h3>Control devices</h3>'
                  '<table><tr><th>Device</th><th>Status</th><th>Actions'
                  '</th></tr>%s</table>' % devrows
                + '</section>')
        if not rows:
            rows = ['<h2>Link your Waveline device</h2>'
                    '<div class="box"><p>No device is linked to this '
                    'account yet.</p>'
                    '<p><button id="gen" class="main">Generate link '
                    'code</button></p>'
                    '<p id="lc" class="hint">The code appears here; type '
                    'it into Waveline: Preferences - Account - Link this '
                    'device.</p></div>'
                    '<script>'
                    'document.getElementById("gen").onclick=function(){'
                    'var x=new XMLHttpRequest();'
                    'x.open("POST","link-code",true);'
                    'x.onload=function(){'
                    'try{var r=JSON.parse(x.responseText);'
                    'document.getElementById("lc").textContent='
                    '"Your code: "+r.code+" (valid 15 minutes, use it '
                    'once)";}'
                    'catch(e){document.getElementById("lc").textContent='
                    '"Could not generate a code - log in again.";}};'
                    'x.send();};'
                    '</script>']
        body = ("""<p class="hint">Logged in as <strong>%(nick)s</strong>
 (%(email)s)</p>
<form method="post" action="logout" style="display:inline">
 <button type="submit" class="secondary">Log out</button></form>
%(rows)s
<p><button id="libstop" class="secondary">Stop library playback</button></p>
<h3>Jukebox (let guests request tracks)</h3>
<div class="box" id="jukebox">
<p id="jukestate" class="hint" role="status" aria-live="polite">Loading...</p>
<p><button id="jukeon" class="main">Open jukebox</button>
<button id="jukeoff" class="secondary">Pause requests</button>
<button id="jukerevoke" class="rec">Close and retire link</button></p>
<div id="jukereq"></div>
</div>
<h3>AudioVault downloads (on the server)</h3>
<div id="avhist" class="hint" role="status" aria-live="polite">Loading...</div>
<p id="devmsg" class="hint" role="status" aria-live="polite"></p>
<script>
function post(path,obj,cb){
 var x=new XMLHttpRequest();
 x.open("POST",path,true);
 x.setRequestHeader("Content-Type","application/json");
 if(cb)x.onload=function(){cb(x)};else x.onload=function(){location.reload()};
 x.send(JSON.stringify(obj));}
Array.prototype.forEach.call(
 document.querySelectorAll("button[data-cmd]"),function(b){
  b.onclick=function(){
   post("cmd",{action:b.getAttribute("data-cmd"),
               pc:b.getAttribute("data-pc")});
   var m=document.getElementById("devmsg");
   if(m)m.textContent="Command sent.";};});
Array.prototype.forEach.call(
 document.querySelectorAll("button[data-ren]"),function(b){
  b.onclick=function(){
   var n=prompt("New name for this device:");
   if(!n)return;
   post("device/rename",{device:b.getAttribute("data-ren"),
       pc:b.getAttribute("data-pc"),name:n});};});
Array.prototype.forEach.call(
 document.querySelectorAll("button[data-ban]"),function(b){
  b.onclick=function(){
   if(!confirm("Unpair this device? It will lose control until it "+
               "pairs again with a fresh code."))return;
   post("device/ban",{device:b.getAttribute("data-ban"),
       pc:b.getAttribute("data-pc")});};});
Array.prototype.forEach.call(
 document.querySelectorAll("button[data-pcren]"),function(b){
  b.onclick=function(){
   var n=prompt("Name this computer (shown in your dashboard):");
   if(!n)return;
   post("pc/rename",{pc:b.getAttribute("data-pcren"),label:n});};});
var player=null,curUrl='';
function ensureAudio(url){
 if(player&&curUrl===url)return player;
 if(player){try{player.pause()}catch(e){}}
 player=new Audio(url);player.preload='none';curUrl=url;
 return player}
function playLib(id,name){
 var a=ensureAudio("lib/"+id);
 a.play().then(function(){
  var m=document.getElementById("devmsg");
  if(m)m.textContent="Playing from library: "+name})
  .catch(function(){
  var m=document.getElementById("devmsg");
  if(m)m.textContent="Could not play - activate the track again"})}
document.getElementById("libstop").onclick=function(){
 if(player){try{player.pause()}catch(e){}}
 var m=document.getElementById("devmsg");
 if(m)m.textContent="Library playback stopped"};
function ytId(u){
 var id='';
 var k=u.indexOf('v=');if(k>=0)id=u.slice(k+2,k+13);
 k=u.indexOf('youtu.be/');if(k>=0)id=u.slice(k+9,k+20);
 k=u.indexOf('shorts/');if(k>=0)id=u.slice(k+7,k+18);
 return /^[A-Za-z0-9_-]{11}$/.test(id)?id:null}
Array.prototype.forEach.call(
 document.querySelectorAll("button[data-cont]"),function(b){
  b.onclick=function(){
   var th=b.getAttribute("data-cont");
   var st=document.querySelector('[data-contst="'+th+'"]');
   var x=new XMLHttpRequest();x.open("POST","status",true);
   x.setRequestHeader("Content-Type","application/json");
   x.onload=function(){try{var r=JSON.parse(x.responseText);
    if(!r.ok)return;
    var pc=null;Array.prototype.forEach.call(r.pcs,function(p){
     if(p.th===th)pc=p});
    if(!pc||!pc.url){
     if(st)st.textContent="Nothing to continue on this PC yet.";
     return}
    var pos=pc.position||0;var name=pc.now_playing||"";
    var f=document.getElementById("ytf");
    if(f)f.remove();
    var yt=ytId(pc.url);
    if(yt){
     var box=document.createElement("div");box.id="ytf";
     var fr=document.createElement("iframe");
     fr.src="https://www.youtube.com/embed/"+yt+
      "?autoplay=1&start="+Math.floor(pos);
     fr.width="100%%";fr.height="260";fr.style.border="0";
     fr.allow="autoplay; encrypted-media";
     box.appendChild(fr);document.body.appendChild(box);
     if(st)st.textContent="Continuing on YouTube: "+name;
     return}
    var url;
    if(pc.url.indexOf("lib:")===0){url="lib/"+pc.url.slice(4)}
    else if(pc.url.indexOf("http")===0){url=pc.url}
    else{if(st)st.textContent="
     This stream type cannot continue in the browser";return}
    var a=ensureAudio(url);
    var at=pc.live?0:pos;
    var p=a.play();
    if(p&&p.then)p.then(function(){
     try{if(at>0)a.currentTime=at}catch(e){}
     if(st)st.textContent=(pc.live?"Joining live: ":"Continuing at "+
      Math.floor(at)+"s: ")+name})
     .catch(function(){
      if(st)st.textContent="Could not play - activate again"})})
    else{
     if(st)st.textContent="Could not play - activate again"}};
   x.send("{}")}});
function libRefresh(){
 var x=new XMLHttpRequest();x.open("POST","lib/list",true);
 x.setRequestHeader("Content-Type","application/json");
 x.onload=function(){try{var r=JSON.parse(x.responseText);if(!r.ok)return;
  Array.prototype.forEach.call(
   document.querySelectorAll("div.lib"),function(box){
    var th=box.getAttribute("data-pc");box.textContent="";
    var mine=r.files.filter(function(f){return f.th===th});
    if(!mine.length){var hint=document.createElement("p");
     hint.className="hint";hint.textContent="No stored music "
      +"for this PC yet. Upload from Waveline: Cloud library.";
     box.appendChild(hint);return}
    mine.forEach(function(f){
     var row=document.createElement("div");
     row.style.display="flex";row.style.gap="0.4rem";
     row.style.marginBottom="0.3rem";
     var b=document.createElement("button");
     b.textContent="\u25B6 "+f.name+" ("+f.mb+" MB)";
     b.style.flex="1";b.style.textAlign="left";b.style.display="block";
     b.onclick=function(){playLib(f.id,f.name)};
     var d=document.createElement("button");d.textContent="\u2715";
     d.setAttribute("aria-label","Remove "+f.name);
     d.onclick=function(){
      post("lib/del",{id:f.id,pc:th},function(){libRefresh()})};
     row.appendChild(b);row.appendChild(d);box.appendChild(row)})
  })}catch(e){}};
 x.send("{}")}
function tick(){
 var x=new XMLHttpRequest();
 x.open("POST","status",true);
 x.setRequestHeader("Content-Type","application/json");
 x.onload=function(){
  try{var r=JSON.parse(x.responseText);if(!r.ok)return;
   Array.prototype.forEach.call(
    document.querySelectorAll("section.pc"),function(sec){
     var th=sec.getAttribute("data-pc");
     var pc=r.pcs.filter(function(p){return p.th===th})[0];
     if(!pc)return;
     var np=sec.querySelector(".np");
     if(np)np.textContent="Now playing: "+(pc.now_playing||
       "(nothing yet - open Waveline)")+" ("+
       (pc.playing?"playing":"paused")+")";
     var st=sec.querySelector("h2 .on, h2 .off");
     if(st){st.className=pc.online?"on":"off";
      st.textContent=pc.online?"online":"offline";}});
  }catch(e){}};
 x.send("{}");}
setInterval(tick,5000);tick();libRefresh();
function avRefresh(){
 var x=new XMLHttpRequest();x.open("POST","av/history",true);
 x.setRequestHeader("Content-Type","application/json");
 x.onload=function(){try{var r=JSON.parse(x.responseText);
  if(!r.ok)return;
  var box=document.getElementById("avhist");
  if(!box)return;box.textContent="";
  if(r.blocked){var w=document.createElement("p");
   w.className="hint";w.textContent="audiovault.net is currently "
    +"challenging the server - new downloads pause and retry "
    +"automatically.";box.appendChild(w)}
  if(!r.rows.length){var e=document.createElement("p");
   e.className="hint";e.textContent="Nothing sent from Waveline yet. "
    +"Open AudioVault in Waveline, pick a title and press "
    +"Send to my devices.";box.appendChild(e);return}
  r.rows.forEach(function(row){
   var p=document.createElement("p");
   var cls=row.status==="done"?"on":(row.status==="failed"?"err":"hint");
   var stat=row.status==="done"?"downloaded - in your cloud library":
    (row.status==="failed"?"failed":
     (row.status==="blocked"?"waiting (site is challenging the server)":
      "waiting to download"));
   var when=new Date((row.updated||0)*1000);
   var whenTxt=isNaN(when.getTime())?"":
    " - "+when.toLocaleString();
   var sp=document.createElement("span");sp.className=cls;
   sp.textContent=stat;
   p.appendChild(sp);
   p.appendChild(document.createTextNode(": "+row.name
    +(row.pc?" ("+row.pc+")":"")+whenTxt));
   if(row.detail){var d=document.createElement("span");
    d.className="hint";d.textContent=" - "+row.detail;p.appendChild(d)}
   box.appendChild(p)})}
  catch(e){}};
 x.send("{}")}
avRefresh();setInterval(avRefresh,10000);
function jukeRefresh(){
 var x=new XMLHttpRequest();x.open("POST","juke/state",true);
 x.setRequestHeader("Content-Type","application/json");
 x.onload=function(){try{var r=JSON.parse(x.responseText);if(!r.ok)return;
  var stp=document.getElementById("jukestate");
  var reqbox=document.getElementById("jukereq");
  var on=document.getElementById("jukeon"),off=document.getElementById("jukeoff");
  if(!stp||!reqbox)return;reqbox.textContent="";
  if(!r.share){stp.textContent="No share yet - open the jukebox to create one.";
   on.disabled=false;off.disabled=true;return}
  stp.textContent=(r.open?"OPEN - share this link: ":"PAUSED - link kept: ")+r.share+
   " ("+r.served+" request(s) served)";
  on.disabled=r.open;off.disabled=!r.open;
  if(r.now&&r.now.name)stp.textContent+=" - playing: "+r.now.name;
  var pending=r.req.filter(function(q){return !q.done});
  var past=r.req.filter(function(q){return q.done});
  if(!pending.length&&!past.length){
   var e=document.createElement("p");e.className="hint";
   e.textContent="No requests yet.";reqbox.appendChild(e);return}
  pending.forEach(function(q){
   var row=document.createElement("div");row.style.margin="0.25rem 0";
   row.textContent=q.name+(q.guest?" (from "+q.guest+")":"")+" - ";
   var yes=document.createElement("button");yes.textContent="Approve";
   yes.onclick=function(){post("juke/decide",{id:q.id,approve:true},
    function(){jukeRefresh()})};
   var no=document.createElement("button");no.className="rec";
   no.textContent="Deny";
   no.onclick=function(){post("juke/decide",{id:q.id,approve:false},
    function(){jukeRefresh()})};
   row.appendChild(yes);row.appendChild(no);reqbox.appendChild(row)});
  past.slice(-8).forEach(function(q){
   var p=document.createElement("p");p.className="hint";
   p.textContent=(q.approved?"Approved: ":"Denied: ")+q.name;
   reqbox.appendChild(p)})}
  catch(e){}};
 x.send("{}")}
document.getElementById("jukeon").onclick=function(){
 post("juke/toggle",{open:true},function(){jukeRefresh()})};
document.getElementById("jukeoff").onclick=function(){
 post("juke/toggle",{open:false},function(){jukeRefresh()})};
document.getElementById("jukerevoke").onclick=function(){
 if(!confirm("Close the jukebox and retire the share link?"))return;
 post("juke/toggle",{revoke:true},function(){jukeRefresh()})};
jukeRefresh();setInterval(jukeRefresh,5000);
</script>""" % {"nick": user["nick"], "email": user["email"],
                "rows": "".join(rows)})
        self._acct_page("Waveline dashboard", "Waveline dashboard", body)

    # -- cross-links: phone pages <-> account dashboard ---------------------

    def _h_acct_linkcode(self):
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        code = self.accounts.issue_link_code(user["email"])
        if not code:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True, "code": code})

    def _linked_pcs(self, email: str):
        """[(token_hash, label), ...] of this account's linked PCs."""
        return self.accounts.linked_full(email)

    def _h_acct_status(self):
        """Poll endpoint for dashboard live refresh (session authed):
        per-PC now-playing, online flag and control-device rows."""
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        pcs = []
        for th, label in self._linked_pcs(user["email"]):
            st = self.store.state_by_th(th)
            devs = self.store.devices_by_th(th)
            pcs.append({
                "th": th,
                "label": label,
                "online": self.store.pc_online(th),
                "now_playing": st.get("now_playing", ""),
                "playing": bool(st.get("playing")),
                # Continue-anywhere (2026-09-25): the dashboard's
                # "Continue on this device" resumes url at position -
                # a lib: id streams from the relay even with the PC off.
                "url": st.get("url", ""),
                "position": float(st.get("position", 0.0) or 0.0),
                "live": bool(st.get("live")),
                "devices": [{"id": d["id"], "name": d["name"],
                             "online": d["online"],
                             "idle_min": d["idle_min"]}
                            for d in devs],
            })
        self._json({"ok": True, "nick": user["nick"], "pcs": pcs})

    def _role_actor(self, token):
        """Pairing token -> account email (token-authed like /app/sync).
        Roles are managed from the APP, so identity rides the pairing
        that is linked to the account (account-only identity model)."""
        th = self._app_auth(token)
        if th is None:
            return None
        email = self._acct_email_for_th(th)
        return email or None

    def _h_acct_roles(self):
        """List accounts+roles (owner/admin, pairing-token authed)."""
        data = self._body()
        actor = self._role_actor((data or {}).get("token"))
        if actor is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        rows = self.accounts.list_users(actor)
        if rows is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True, "role": self.accounts.role_of(actor),
                    "users": [{"email": e, "nick": n, "role": r,
                               "verified": v}
                              for (e, n, r, v) in rows]})

    def _h_acct_role_set(self):
        """Assign a role (owner: any; admin: helper/user only)."""
        data = self._body()
        actor = self._role_actor((data or {}).get("token"))
        if actor is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        ok, problem = self.accounts.set_role(
            actor, str(data.get("email", "")),
            str(data.get("role", "")))
        if not ok:
            return self._json({"ok": False, "error": problem}, 403)
        # Tell the target about their new role in-app (next sync).
        try:
            self.accounts.notify_user(
                str(data.get("email", "")), "role",
                "Your role is now %s" % str(data.get("role", "")))
        except Exception:
            pass
        self._json({"ok": True})

    def _h_feedback_list(self):
        """Report summaries for feedback.read roles (token authed)."""
        data = self._body()
        actor = self._role_actor((data or {}).get("token"))
        if actor is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        rows = self.accounts.feedback_list(actor)
        if rows is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True, "reports": rows})

    def _h_feedback_get(self):
        """One full report (message + redacted log) for readers."""
        data = self._body()
        actor = self._role_actor((data or {}).get("token"))
        if actor is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        report = self.accounts.feedback_get(
            actor, str((data or {}).get("ticket", "")))
        if report is None:
            return self._json({"ok": False, "error": "not found"}, 404)
        self._json({"ok": True, "report": report})

    def _h_acct_pc_rename(self):
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = str(data.get("pc", ""))
        label = str(data.get("label", ""))[:40]
        if not th or not label.strip():
            return self._json({"ok": False, "error": "bad request"}, 400)
        # only PCs actually linked to THIS account may be renamed
        if th not in [t for t, _l in self._linked_pcs(user["email"])]:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": self.accounts.rename_pc(
            user["email"], th, label)})

    def queue_pair_alert_if_new(self, ref: str, devid: str,
                                name: str) -> None:
        """After a phone cmd/state touched its pairing: if THAT touch
        registered a new control device, alert the linked account's
        owner.  No-op for repeat connections (pop_just_paired).
        *ref* is the code-or-TH the phone used (TH on account-only
        sessions - the alert text shows a stable short form of it)."""
        try:
            if self.TH_RE.match(str(ref or "")):
                th = str(ref)[:64]
            else:
                th = self.store.th_for(ref)
            if not th:
                return
            new_id = self.store.pop_just_paired(th)
            if not new_id or new_id != str(devid)[:64]:
                return
            email = self._acct_email_for_th(th)
            if email:
                short = str(ref)[:10]
                self.queue_alert("pair", email,
                                 "control device '%s' paired (%s)"
                                 % (str(name or "").strip()[:40]
                                    or "unnamed", short))
        except Exception:
            pass

    def _h_acct_dash_cmd(self):
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        action = str(data.get("action", ""))
        if action not in VALID_ACTIONS:
            return self._json({"ok": False, "error": "bad request"}, 400)
        results = []
        target = str(data.get("pc", ""))  # optional: one specific PC
        linked = self._linked_pcs(user["email"])
        targets = ([(t, l) for t, l in linked if t == target] if target
                   else linked)
        for th, _label in targets:
            results.append(self.store.push_by_th(th, action))
        self._json({"ok": bool(results) and all(r == "ok" for r in results),
                    "sent": len(results)})

    def _h_acct_lib_list(self):
        """Cloud Library listing for the dashboard (session authed):
        every linked PC's stored uploads, so the owner can browse and
        play them in the browser - including with the PC switched off."""
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        files = []
        for th, label in self._linked_pcs(user["email"]):
            for l in self.store.lib_list(th):
                files.append({"id": l["id"], "name": l["name"],
                              "mb": round(l["bytes"] / 1e6, 1),
                              "pc": label, "th": th})
        self._json({"ok": True, "files": files})

    def _h_acct_av_history(self):
        """AudioVault server-download history for the dashboard (session
        authed): pending / done / failed / blocked jobs across all of
        this account's linked PCs, newest first."""
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        emails = [user["email"]] + [e for e, _t in
                                    self.accounts.linked_full_emails()
                                    if e == user["email"]]
        rows = self.av_worker.history_for(emails) if self.av_worker \
            else []
        labels = dict(self._linked_pcs(user["email"]))
        for row in rows:
            row["pc"] = labels.get(row.get("th", ""), "")
        # A blocked backoff is worker-global; surface it once.
        blocked = bool(getattr(self.av_worker, "_blocked_until", 0)
                       > time.time()) if self.av_worker else False
        self._json({"ok": True, "rows": rows, "blocked": blocked})

    def _h_acct_lib_del(self):
        """Dashboard-side delete of one stored library file (session
        authed; the pc token hash must belong to THIS account)."""
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        lid = str(data.get("id", ""))
        th = str(data.get("pc", ""))
        if not self.LIB_RE.match(lid) or \
                th not in [t for t, _l in self._linked_pcs(user["email"])]:
            return self._json({"ok": False, "error": "bad request"}, 400)
        self._json({"ok": self.store.lib_del(th, lid)})

    # -- audiobook bookmarks + listening digest (entry 49) ------------------
    def _h_lib_stream(self):
        """Token-authed Range stream of one cloud-library track - the
        account bookmark resume path when the local copy is missing."""
        from urllib.parse import parse_qs, urlparse
        try:
            q = parse_qs(urlparse(self.path).query)
            token = str(q.get("token", [""])[0])
            lid = str(q.get("lid", [""])[0])
        except Exception:
            return self._reject()
        if not self.cmd_rate.allow(self._ip()):
            return self._throttled()
        th = self._app_auth(token)
        if th is None or not self.LIB_RE.match(lid):
            return self._reject()
        l = self.store.lib_get(th, lid)
        if l is None or not os.path.isfile(l["file"]):
            return self._json({"ok": False, "error": "missing"}, 404)
        return self._send_audio(l)

    def _h_bm_put(self):
        """App-side bookmark sync (token authed): one named position on
        any playable URL.  The account is derived from the token, so
        every linked PC shares the bookmark set."""
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        email = self._acct_email_for_th(th)
        if not email:
            return self._json({"ok": False,
                               "error": "no account linked"}, 400)
        url = str(data.get("url", ""))
        if not url:
            return self._json({"ok": False, "error": "bad request"}, 400)
        bid = hashlib.sha256(url.lower().encode("utf-8")).hexdigest()[:32]
        self.store.bookmark_put(email, bid, url, data.get("name", ""),
                                data.get("pos", 0.0), data.get("label", ""))
        self._json({"ok": True, "id": bid})

    def _h_bm_list(self):
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        email = self._acct_email_for_th(th)
        rows = [{"id": k, **v} for k, v in
                self.store.bookmarks(email)] if email else []
        self._json({"ok": True, "bookmarks": rows})

    def _h_bm_del(self):
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        email = self._acct_email_for_th(th)
        ok = bool(email) and self.store.bookmark_del(
            email, str(data.get("id", "")))
        self._json({"ok": ok})

    def _h_pos_put(self):
        """Cross-device playback position (token authed): upsert one
        track position for the account (6.0 'continue anywhere')."""
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        email = self._acct_email_for_th(th)
        if not email:
            return self._json({"ok": False,
                               "error": "no account linked"}, 400)
        track = str(data.get("track", ""))
        if not track:
            return self._json({"ok": False, "error": "bad request"}, 400)
        self.store.position_put(email, track, data.get("pos", 0.0),
                                data.get("device", ""))
        self._json({"ok": True})

    def _h_pos_list(self):
        """Most recent account-wide positions (token authed)."""
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        email = self._acct_email_for_th(th)
        rows = [{"track": k, **v} for k, v in
                self.store.positions(email)] if email else []
        rows.sort(key=lambda r: -float(r.get("at") or 0.0))
        self._json({"ok": True, "positions": rows[:50]})

    def _h_stats_note(self):
        """App-side listening-time accumulator (token authed): the app
        reports per-track seconds on sync; powers the weekly digest."""
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        email = self._acct_email_for_th(th)
        if not email:
            return self._json({"ok": True})   # not linked: nothing to do
        try:
            secs = min(3600.0, max(0.0, float(data.get("secs", 0))))
        except (TypeError, ValueError):
            secs = 0.0
        self.store.listen_add(email, data.get("name", ""), secs)
        self._json({"ok": True})

    # -- jukebox: owner (dashboard) endpoints ------------------------------
    def _h_juke_state(self):
        """Dashboard poll: the account's share + request queue."""
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        email = user["email"]
        j = self.store.juke_get(email)
        state = {"ok": True, "open": False, "share": "", "req": [],
                 "now": {}, "served": 0}
        if j:
            state.update({
                "open": bool(j.get("open")),
                "share": "%s/juke/%s" % (_accounts.Mailer().public_url,
                                          j.get("tok", "")),
                "now": dict(j.get("now", {})),
                "served": int(j.get("reqs_served", 0)),
                "req": [{"id": k, "name": v.get("name", ""),
                         "guest": v.get("guest", ""),
                         "done": bool(v.get("done")),
                         "approved": bool(v.get("approved"))}
                        for k, v in self.store.juke_requests(email)],
            })
        self._json(state)

    def _h_juke_toggle(self):
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        email = user["email"]
        if data.get("revoke"):
            self.store.juke_revoke(email)
            return self._json({"ok": True})
        if not self.store.juke_get(email):
            self.store.juke_create(email)
        self.store.juke_set_open(email, bool(data.get("open")))
        self._json({"ok": True})

    def _h_juke_decide(self):
        """Owner approves/denies a request; approval queues it on the
        linked PCs and (when idle) starts it immediately."""
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        email = user["email"]
        row = self.store.juke_decide(email, data.get("id", ""),
                                     bool(data.get("approve")))
        if not row:
            return self._json({"ok": False, "error": "bad request"}, 400)
        if row.get("approved"):
            for th, _label in self._linked_pcs(email):
                if self.store.pc_online(th):
                    self.store.push_by_th(th, "play_pause")
                    break
        self._json({"ok": True})

    # -- jukebox: guest (share-link) endpoints ------------------------------
    def _h_juke_guest_page(self, jtok):
        """Public request page for a live share (token in the URL)."""
        email, j = self.store.juke_by_token(jtok)
        if not email or not j.get("open"):
            body = ("<h1>Jukebox not available</h1>"
                    "<p>This share was closed by its owner.</p>")
            return self._acct_page("Waveline jukebox", "Waveline jukebox",
                                   body)
        nick = self.accounts.nick_for(email) if self.accounts else ""
        body = ("<p>Request a track from <strong>%s</strong>'s Waveline "
                "library. The owner approves requests before they "
                "play.</p>"
                "<div id=\"lib\" class=\"hint\" role=\"status\" "
                "aria-live=\"polite\">Loading...</div>"
                "<p id=\"msg\" role=\"status\" aria-live=\"polite\"></p>"
                "<script>\nvar TOK=%s;\n%s\n</script>"
                % (nick or "your host", json.dumps(jtok),
                   self._JUKE_GUEST_JS))
        self._acct_page("Waveline jukebox", "Waveline jukebox", body)

    _JUKE_GUEST_JS = r"""
function post(path,obj,cb){
 var x=new XMLHttpRequest();x.open("POST",path,true);
 x.setRequestHeader("Content-Type","application/json");
 if(cb)x.onload=function(){cb(x)};x.send(JSON.stringify(obj));}
post("list",{},function(x){
 try{var r=JSON.parse(x.responseText);if(!r.ok)return;
  var box=document.getElementById("lib");box.textContent="";
  if(!r.files.length){box.textContent="The library is empty right now.";
   return}
  r.files.forEach(function(f){
   var b=document.createElement("button");
   b.textContent="Request "+f.name;
   b.style.display="block";b.style.margin="0.25rem 0";
   b.onclick=function(){
    post("req",{lid:f.id,name:f.name},function(x2){
     try{var r2=JSON.parse(x2.responseText);
      var m=document.getElementById("msg");
      if(m)m.textContent=r2.ok?"Request sent - waiting for the owner":
       (r2.error==="dupe"?"Already requested":
        (r2.error==="closed"?"Requests are closed":"Could not send"));
     }catch(e){}});};
   box.appendChild(b)})}
 catch(e){document.getElementById("lib").textContent=
  "Could not load the library."});});
"""

    def _h_juke_guest_list(self, jtok):
        email, j = self.store.juke_by_token(jtok)
        if not email or not j.get("open"):
            return self._json({"ok": False, "error": "closed"}, 404)
        files = []
        for th, label in self._linked_pcs(email):
            for l in self.store.lib_list(th):
                files.append({"id": l["id"], "name": l["name"]})
        self._json({"ok": True, "files": files[:200]})

    def _h_juke_guest_req(self, jtok):
        email, j = self.store.juke_by_token(jtok)
        if not email or not j.get("open"):
            return self._json({"ok": False, "error": "closed"}, 404)
        data = self._body()
        lid = str(data.get("lid", ""))
        if not data or not self.LIB_RE.match(lid):
            return self._json({"ok": False, "error": "bad request"}, 400)
        name = str(data.get("name", "track"))[:MAX_STR]
        rid = secrets.token_urlsafe(9)
        verdict = self.store.juke_request(email, rid, lid, name,
                                          data.get("guest", "guest"))
        if verdict != "ok":
            return self._json({"ok": False, "error": verdict}, 400)
        self._json({"ok": True})

    def _h_acct_device_rename(self):
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        devid = str(data.get("device", ""))[:64]
        name = str(data.get("name", ""))[:40]
        if not devid or not name.strip():
            return self._json({"ok": False, "error": "bad request"}, 400)
        verdicts = [self.store.rename_device(t, devid, name)
                    for t, _l in self._linked_pcs(user["email"])]
        self._json({"ok": "ok" in verdicts})

    def _h_acct_device_ban(self):
        user = self._acct_user()
        if user is None:
            return self._json({"ok": False, "error": "forbidden"}, 403)
        data = self._body()
        if not data:
            return self._json({"ok": False, "error": "bad request"}, 400)
        devid = str(data.get("device", ""))[:64]
        if not devid:
            return self._json({"ok": False, "error": "bad request"}, 400)
        n = sum(self.store.ban_device_by_th(t, devid)
                for t, _l in self._linked_pcs(user["email"]))
        self._json({"ok": True, "banned": n})

    # -- routing -----------------------------------------------------------
    # Landing page: user enters the code - no URL surgery needed.
    _PHONE_LANDING = r"""<!doctype html><html><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Waveline remote</title>
<style>
body{font-family:system-ui,sans-serif;background:#0f1319;color:#e8eaed;
 max-width:30rem;margin:6vh auto;padding:0 1.2rem;text-align:center}
h1{font-size:1.7em;margin-bottom:.2em}
p.sub{color:#9aa4b0;margin-top:0}
.box{background:#171d26;border-radius:14px;padding:1.2rem 1rem;
 margin-top:1rem}
button{font-size:1.05em;background:#1f2937;color:#e8eaed;border:0;
 border-radius:10px;padding:.55em 1em;margin:.2rem}
button.main{background:#4a90d9;font-size:1.25em;padding:.6em 1.6em}
#err{color:#ff8a80;min-height:1.2em}
.hint{color:#9aa4b0;font-size:.9em}
</style>
</head><body>
<h1>Waveline Remote</h1>
<p class="sub">Control your music, radio and podcasts from anywhere.</p>
<div class="box">
 <p><button id="acct" class="main"
  onclick="location='account/'">Open my Waveline</button></p>
 <p id="err" aria-live="polite"></p>
 <p class="hint">Sign in with your Waveline account to control your
  computers, listen to your stored music and see every device.</p>
</div>
<p class="hint">No account yet? Create one free in the Waveline app
 (Preferences - Account) and link this computer - it then appears in
 your <a href="account" style="color:#4a90d9">account dashboard</a> on
 every device.</p>
<script></script></body></html>"""

    # Identical body for every unknown/expired/used code: no oracle.
    _PHONE_MISS = """<!doctype html><html><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Waveline remote</title>
<style>
body{font-family:system-ui,sans-serif;background:#0f1319;color:#e8eaed;
 max-width:30rem;margin:6vh auto;padding:0 1.2rem;text-align:center}
h1{font-size:1.7em;margin-bottom:.2em}
p.sub{color:#9aa4b0;margin-top:0}
.box{background:#171d26;border-radius:14px;padding:1.2rem 1rem;
 margin-top:1rem}
button{font-size:1.05em;background:#1f2937;color:#e8eaed;border:0;
 border-radius:10px;padding:.55em 1em;margin:.2rem}
button.main{background:#4a90d9;font-size:1.25em;padding:.6em 1.6em}
button.rec{background:#7a3b3b}
#np{font-size:1.15em;font-weight:600;min-height:1.4em}
#vol,#st{color:#9aa4b0;min-height:1.2em;margin:.2rem 0}
#err{color:#ff8a80;min-height:1.2em}
.hint{color:#9aa4b0;font-size:.9em}
input#c{font-size:1.9em;letter-spacing:.35em;width:12em;text-align:center;
 text-transform:uppercase;background:#0f1319;color:#4a90d9;
 border:2px solid #2a3644;border-radius:10px;padding:.25em 0}
</style>
</head><body>
<h1>Pairing not found</h1>
<p>This link is no longer valid.  Waveline is account-based: sign in
 with the owner's Waveline account to control their computers - no
 codes to type.</p>
<p><a href="../account" style="color:#4a90d9">Open my Waveline</a> &#183;
 <a href="../" style="color:#4a90d9">Waveline remote home</a></p>
</body></html>"""

    _PHONE_PAGE = r"""<!doctype html><html><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Waveline remote</title>
<style>
body{font-family:system-ui,sans-serif;background:#0f1319;color:#e8eaed;
 max-width:30rem;margin:6vh auto;padding:0 1.2rem;text-align:center}
h1{font-size:1.7em;margin-bottom:.2em}
p.sub{color:#9aa4b0;margin-top:0}
.box{background:#171d26;border-radius:14px;padding:1.2rem 1rem;
 margin-top:1rem}
button{font-size:1.05em;background:#1f2937;color:#e8eaed;border:0;
 border-radius:10px;padding:.55em 1em;margin:.2rem}
button.main{background:#4a90d9;font-size:1.25em;padding:.6em 1.6em}
button.rec{background:#7a3b3b}
#np{font-size:1.15em;font-weight:600;min-height:1.4em}
#vol,#st{color:#9aa4b0;min-height:1.2em;margin:.2rem 0}
#err{color:#ff8a80;min-height:1.2em}
.hint{color:#9aa4b0;font-size:.9em}
input#c{font-size:1.9em;letter-spacing:.35em;width:12em;text-align:center;
 text-transform:uppercase;background:#0f1319;color:#4a90d9;
 border:2px solid #2a3644;border-radius:10px;padding:.25em 0}
</style>
</head><body>
<h1>Waveline Remote</h1>
<p class="sub">Paired - this device can control Waveline until the
 owner revokes it. Keep this page open, or save it to your home
 screen.</p>
<div class="box">
 <p id="np" aria-live="polite">Not playing</p>
 <p id="vol"></p>
 <p>
  <button onclick="c(&quot;prev_track&quot;)">&#9198;</button>
  <button class="main" onclick="c(&quot;play_pause&quot;)">&#9654; / &#10074;&#10074;</button>
  <button onclick="c(&quot;next_track&quot;)">&#9197;</button>
  <button onclick="c(&quot;stop&quot;)">&#9632;</button>
 </p>
 <p>
  <button onclick="c(&quot;seek_back&quot;)">&#8210;10s</button>
  <button onclick="c(&quot;seek_forward&quot;)">+10s</button>
 </p>
 <p>
  <button onclick="c(&quot;volume_down&quot;)">Vol &#8210;</button>
  <button onclick="c(&quot;mute&quot;)">Mute</button>
  <button onclick="c(&quot;volume_up&quot;)">Vol +</button>
 </p>
</div>
<div class="box" id="alarmbox" style="display:none">
 <p class="hint" id="alarmname">Alarm</p>
 <p>
  <button class="main" onclick="c(&quot;alarm_snooze&quot;)">&#9200; Snooze</button>
  <button class="rec" onclick="c(&quot;alarm_dismiss&quot;)">&#10060; Dismiss</button>
 </p>
 <p class="hint">Snooze stops it now and starts the same alarm again
  after its snooze minutes. Dismiss silences it for good.</p>
</div>
<div class="box">
 <p class="hint">Sound</p>
 <p>
  <button onclick="c(&quot;bass_down&quot;)">Bass &#8210;</button>
  <button onclick="c(&quot;bass_up&quot;)">Bass +</button>
  <button onclick="c(&quot;treble_down&quot;)">Treble &#8210;</button>
  <button onclick="c(&quot;treble_up&quot;)">Treble +</button>
  <button onclick="c(&quot;tone_reset&quot;)">Tone reset</button>
 </p>
 <p class="hint">Playback</p>
 <p>
  <button onclick="c(&quot;speed_down&quot;)">Speed &#8210;</button>
  <button onclick="c(&quot;speed_reset&quot;)">Speed 1x</button>
  <button onclick="c(&quot;speed_up&quot;)">Speed +</button>
  <button onclick="c(&quot;toggle_shuffle&quot;)">Shuffle</button>
  <button onclick="c(&quot;cycle_repeat&quot;)">Repeat</button>
  <button onclick="c(&quot;stop_after_current&quot;)">Stop after</button>
 </p>
 <p class="hint">Extras</p>
 <p>
  <button onclick="c(&quot;chapter_prev&quot;)">Chapter &#8210;</button>
  <button onclick="c(&quot;chapter_next&quot;)">Chapter +</button>
  <button onclick="c(&quot;bookmark_add&quot;)">Bookmark</button>
  <button onclick="c(&quot;announce_time&quot;)">Say time</button>
  <button onclick="c(&quot;sleep_timer&quot;)">Sleep timer</button>
 </p>
 <p>
  <button class="rec" onclick="c(&quot;record_now&quot;)">&#9679; Record</button>
  <button class="rec" onclick="c(&quot;stop_recording&quot;)">&#9632; Stop rec</button>
 </p>
</div>
<div class="box">
 <p class="hint">Listen on this device</p>
 <p id="lap" aria-live="polite" style="min-height:1.2em;color:#9aa4b0"></p>
 <p>
  <button id="listenbtn" class="main"
   onclick="listen()">&#127911; Play here</button>
  <button id="stopbtn" onclick="stopListen()">Stop</button>
 </p>
 <p class="hint">Plays the same music as the PC. Radio keeps playing
  live while the PC is on; uploaded tracks also work when it is off.</p>
 <div id="libbox" style="display:none;text-align:left">
  <p class="hint">Stored music (works with the PC off)</p>
  <div id="lib"></div>
 </div>
</div>
<p id="err" aria-live="polite"></p>
<p class="hint">Waveline remote &#183; paired via your-domain
 &#183; <a href="../account" style="color:#4a90d9">Account dashboard</a>
 (see all your devices and control your player from anywhere)</p>
<script>
var BASE=location.pathname.replace(/\/$/, '');
var CODE=BASE.split('/').pop();
var DKEY='wv_dev_'+CODE;
var dev=JSON.parse(localStorage.getItem(DKEY)||'null');
if(!dev||!dev.id){
 var b=new Uint8Array(16);crypto.getRandomValues(b);
 var id='';for(var i=0;i<b.length;i++)id+=('0'+b[i].toString(16)).slice(-2);
 var name=prompt('Name this device (shown to the owner):','My device')
  ||'My device';
 dev={id:id,name:String(name).slice(0,40)};
 localStorage.setItem(DKEY,JSON.stringify(dev));}
function h(){return {'Content-Type':'application/json',
 'X-Device-Id':dev.id,'X-Device-Name':dev.name}}
function removed(){document.getElementById("err").textContent=
 "This device was removed by the owner - ask for a new pairing"}
function closed(){document.getElementById("err").textContent=
 "This computer is no longer shared with this browser. Open it from "
 + "your account again"}
function c(a){fetch(BASE+'/cmd',{method:'POST',headers:h(),
 body:JSON.stringify({action:a})}).then(r=>{if(r.status===403)
 r.json().then(j=>{(j.error==='pairing-closed')?closed():removed()})})
 .catch(()=>{})}
function set(id,t){var el=document.getElementById(id);
 if(el.textContent!==t)el.textContent=t}
function ytId(u){var m=u.match(/(?:v=|youtu\.be\/|shorts\/)([\w-]{11})/);
 return m?m[1]:null}
var player=null,curUrl='',liveOnly=false;
function ensureAudio(url){
 if(player&&curUrl===url)return player;
 if(player){try{player.pause()}catch(e){}}
 player=new Audio(url);player.preload='none';curUrl=url;
 return player}
function playLib(id,name){
 var a=ensureAudio(BASE+'/lib/'+id);
 a.play().then(()=>{set('lap','Playing from library: '+name);
  liveOnly=false;libRefresh()}).catch(()=>{
  set('lap','Could not play - tap again (browsers need a tap)')})}
function listen(){
 fetch(BASE+'/state',{headers:h()}).then(r=>r.json()).then(j=>{
  var s=j.state||{};var u=s.url||'';
  if(!u){set('lap','Nothing playing on the PC right now');return}
  if(u.indexOf('lib:')===0){
   playLib(u.slice(4),s.now_playing||'track');return}
  if(s.live){
   var a=ensureAudio(u);a.play().then(()=>{
    set('lap','Playing radio (live edge): '+ (s.now_playing||''));
    liveOnly=true;libRefresh()}).catch(()=>set('lap','Could not start radio'));
   return}
  if(ytId(u)){
   // YouTube: browsers block direct streams - use an embed
   var box=document.getElementById('libbox');
   var f=document.createElement('iframe');
   f.src='https://www.youtube.com/embed/'+ytId(u)+'?autoplay=1';
   f.width='100%';f.height='200';f.style.border='0';
   f.allow='autoplay; encrypted-media';
   var old=document.getElementById('ytf');
   if(old)old.remove();f.id='ytf';
   box.style.display='block';box.insertBefore(f,box.firstChild);
   set('lap','Playing YouTube on this device (separate stream)');
   liveOnly=false;libRefresh();return}
  var a=ensureAudio(u);a.play().then(()=>{
   set('lap','Playing: '+(s.now_playing||u));
   liveOnly=false;libRefresh()}).catch(()=>
   set('lap','This stream cannot play in the browser'));
 }).catch(()=>set('lap','Could not reach the relay'))}
function stopListen(){
 if(player){try{player.pause()}catch(e){}}
 var f=document.getElementById('ytf');if(f)f.remove();
 set('lap','');libRefresh()}
function libRefresh(){
 fetch(BASE+'/liblist',{method:'POST',headers:h(),
  body:JSON.stringify({code:CODE,device_id:dev.id})}).then(r=>r.json()).then(j=>{
  var files=(j&&j.files)||[];var box=document.getElementById('libbox');
  var el=document.getElementById('lib');el.textContent='';
  if(!files.length&&liveOnly===false){box.style.display='none';return}
  box.style.display='block';
  files.forEach(function(f){
   var b=document.createElement('button');
   b.textContent='\u25B6 '+f.name+' ('+f.mb+' MB)';
   b.style.display='block';b.style.width='100%';b.style.textAlign='left';
   b.onclick=function(){playLib(f.id,f.name)};
   var d=document.createElement('button');
   d.textContent='\u2715';d.setAttribute('aria-label','Remove '+f.name);
   d.onclick=function(){
    fetch(BASE+'/libdel',{method:'POST',headers:h(),
     body:JSON.stringify({code:CODE,device_id:dev.id,id:f.id})}).then(()=>libRefresh())};
   var row=document.createElement('div');
   row.style.display='flex';row.style.gap='0.4rem';
   row.style.marginBottom='0.3rem';
   row.appendChild(b);row.appendChild(d);el.appendChild(row)})
 }).catch(()=>{})}
libRefresh();
function poll(){fetch(BASE+'/state',{headers:h()}).then(r=>{
 if(r.status===404){
  set("err","Pairing not found - the code may be wrong, expired, or already used");throw 0}
 if(r.status===403){
  r.json().then(j=>{(j.error==='pairing-closed')?closed():removed()});
  throw 0}
 return r.json()}).then(j=>{
 var s=j.state||{};
 var pend=(s.volume==null&&!s.now_playing);
 var t=pend?'Waiting for the owner to start Waveline…'
       :(s.now_playing?s.now_playing+(s.playing?'':' (paused)'):'Not playing');
 set("np",t);
 var ab=document.getElementById('alarmbox');
 if(s.alarm_active){ab.style.display='block';
  set("alarmname",'Alarm: '+(s.alarm_label||'ringing'));}
 else ab.style.display='none';
 set("vol",pend?'':(s.volume!=null?('Volume: '+Math.round(s.volume)+'%'):''));
 set("err","");
 }).catch(()=>{})}
setInterval(poll,2000);poll();
</script></body></html>"""

    def _h_acct_remote_page(self, ident):
        """The PHONE PAGE served under the OWNER's session (2026-09-25:
        phones use the account now - pairing codes are app-internal).
        *ident* is a linked PC's token hash (/account/remote/<th>) or a
        pairing code (legacy /c/<code> with a session); __CODE__ in the
        page template is ALWAYS filled with the token hash and every
        phone fetch then authenticates by session cookie."""
        user = self._acct_user()
        if user is None:
            return self._acct_page(
                "Waveline account", "Please log in",
                self._msg(False, "Your session expired - log in again.")
                + self._h_acct_login_form_only())
        if self.TH_RE.match(ident or ""):
            th = str(ident)[:64]
        else:
            th = ""
        if not th or th not in [t for t, _l in
                                self._linked_pcs(user["email"])]:
            return self._h_phone_miss()
        # The shared template's footer link is dir-relative for /c/<code>;
        # from /account/remote/<th> it must climb one level less.
        body = self._PHONE_PAGE.replace("__CODE__", th)\
            .replace('href="../account"', 'href="../dashboard"')\
            .encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    TH_RE = re.compile(r"^[0-9a-f]{64}$")

    def _ensure_account_pc(self, code: str) -> str:
        """Token hash of the PC this request may touch, when it carries a
        live account session linked to that PC; "" otherwise.  *code* is
        the PC's token hash (the session phone page /account/remote/<th>
        fills its fetches with the hash) - the phone endpoints then
        authenticate by session cookie INSTEAD of device headers."""
        if self.accounts is None:
            return ""
        user = self._acct_user()
        if user is None:
            return ""
        if not self.TH_RE.match(code or ""):
            return ""
        th = str(code)[:64]
        if not th:
            return ""
        return th if th in [t for t, _l in
                            self._linked_pcs(user["email"])] else ""

    @classmethod
    def _pc_ref_ok(cls, code: str) -> bool:
        """Guard for phone endpoints: the ref is a session token hash."""
        return bool(cls.TH_RE.match(code or ""))

    def _h_phone_page(self, code):
        # Legacy /c/<code> entry page: codes are gone, so every lookup
        # misses with the SAME friendly page (no oracle).  The route is
        # kept only so old bookmarks get the account-first hint instead
        # of a bare 404.
        return self._h_phone_miss()

    def _h_phone_page_unused(self, code):
        body = self._PHONE_PAGE.replace("__CODE__", code).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _h_phone_miss(self):
        body = self._PHONE_MISS.encode("utf-8")
        self.send_response(404)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    # -- cloud library -------------------------------------------------------
    LIB_RE = _LIB_RE
    AUDIO_TYPES = {".mp3": "audio/mpeg", ".m4a": "audio/mp4",
                   ".ogg": "audio/ogg", ".oga": "audio/ogg",
                   ".opus": "audio/ogg", ".wav": "audio/wav",
                   ".flac": "audio/flac", ".webm": "audio/webm"}

    def _audio_type(self, name: str) -> str:
        ext = os.path.splitext(name)[1].lower()
        return self.AUDIO_TYPES.get(ext, "audio/mpeg")

    def _h_lib_list(self):
        data = self._body()
        if data is None:
            return self._reject()
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        rows = [{"id": l["id"], "name": l["name"],
                 "mb": round(l["bytes"] / 1e6, 1), "added": l["added"]}
                for l in self.store.lib_list(th)]
        self._json({"ok": True, "files": rows})

    def _h_lib_del(self):
        data = self._body()
        if data is None:
            return self._reject()
        lid = str(data.get("id", ""))
        th = self._app_auth(data.get("token"))
        if not self.LIB_RE.match(lid) or th is None:
            return self._reject()
        if not self.store.lib_del(th, lid):
            return self._json({"ok": False, "error": "missing"}, 404)
        self._json({"ok": True})

    def _h_lib_upload(self):
        """App pushes a file (<= 2 GB) into the session library.
        Auth: code+token headers; the body is raw audio bytes."""
        # Dedicated lib_rate: uploads are the most expensive operation on
        # the relay (disk I/O) and must be bounded even when authed - but
        # generous, because a legit FOLDER upload can be 200 files in a
        # row (sharing cmd_rate would throttle real users mid-batch).
        if not self.lib_rate.allow(self._ip()):
            return self._throttled()
        # ACCOUNT-ONLY identity: the app proves itself with its token
        # alone (the pairing code field of older clients is ignored).
        token = str(self.headers.get("X-WL-Token", ""))
        lid = str(self.headers.get("X-WL-Lid", ""))
        name = str(self.headers.get("X-WL-Name", "track"))
        # New clients send the name's utf-8 bytes re-mapped through
        # latin-1 (headers cannot carry non-latin1 text; the old code
        # stored mojibake for any emoji/CJK name).  The try/except keeps
        # current deployed clients working unchanged.
        try:
            name = name.encode("latin-1").decode("utf-8")
        except (UnicodeEncodeError, UnicodeDecodeError):
            pass
        name = name[:MAX_STR]
        if not TOKEN_RE.match(token) or not self.LIB_RE.match(lid):
            return self._reject()
        th = self._app_auth(token)
        if th is None:
            return self._reject()
        try:
            size = int(self.headers.get("Content-Length", "0") or "0")
        except ValueError:
            return self._json({"ok": False, "error": "size"}, 400)
        if size <= 0 or size > self.store.LIB_MAX_FILE:
            return self._json({"ok": False, "error": "size"}, 413)
        add_result = self.store.lib_add(th, lid, name, size)
        if add_result == "disk":
            return self._json({"ok": False, "error": "disk"}, 507)
        if add_result != "ok":
            return self._json({"ok": False, "error": "quota"}, 507)
        path = self.store._lib_file(th, lid)
        try:
            os.makedirs(os.path.dirname(path), exist_ok=True)
            left = size
            sha = hashlib.sha256()
            with open(path, "wb") as f:
                while left > 0:
                    chunk = self.rfile.read(min(65536, left))
                    if not chunk:
                        break
                    f.write(chunk)
                    sha.update(chunk)
                    left -= len(chunk)
        except Exception:
            self.store.lib_del(th, lid)
            return self._json({"ok": False, "error": "io"}, 500)
        if left != 0:
            self.store.lib_del(th, lid)
            return self._json({"ok": False, "error": "truncated"}, 400)
        self._json({"ok": True, "sha256": sha.hexdigest()})

    def _h_av_enqueue(self):
        """The PC's AudioVault dialog asks the VPS to download a title
        into this session's Cloud Library.  Auth: code+token (same tier
        as the other /app/lib routes).  Body: {url, name}.  The job is
        downloaded later by _AudioVaultWorker with the server's own
        session - the owner's credentials never travel here."""
        if not self.av_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._reject()
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        url = str(data.get("url", ""))
        # Only audiovault.net title URLs are ever fetched - this endpoint
        # cannot become an SSRF gadget.
        if not url.startswith("https://audiovault.net/"):
            return self._json({"ok": False, "error": "url"}, 400)
        name = str(data.get("name", ""))[:AV_MAX_TITLE]
        email = self._acct_email_for_th(th)
        result = self.av_worker.enqueue(th, email, url, name,
                                        requester=th[:12])
        if result == "ok":
            return self._json({"ok": True,
                               "note": self.av_worker._note_for(th)})
        return self._json({"ok": False, "error": result}, 409 if
                          result == "quota" else 404)

    def _h_av_status(self):
        """How many AudioVault jobs are queued/active for this session."""
        if not self.av_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._reject()
        th = self._app_auth(data.get("token"))
        if th is None:
            return self._reject()
        return self._json({"ok": True, "status": self.av_worker.status(th)})

    # -- VPS agent (server-side downloads / recordings / cookies) -----------
    def _agent_auth(self, cap: int = MAX_BODY, size_error: bool = False):
        """Shared auth + worker check for the /app/agent routes.
        Returns (th, email, data) or None after answering the request.
        The email ("" when the pairing is not linked to an account) is
        the agent's per-user key; features that need persistence tell
        the user to link an account first.  `cap` raises the body-size
        ceiling for routes that carry big payloads (cookies put); with
        size_error=True an over-cap body answers an honest 413 "size"
        instead of the generic reject (which made a real cookies.txt
        look like a bad token - the 2026-10-02 bug)."""
        if not self.agent_rate.allow(self._ip()):
            self._throttled()
            return None
        data = self._body_with_limit(cap)
        if data == "too large":
            if size_error:
                self._json({"ok": False, "error": "size"}, 413)
            else:
                self._reject()
            return None
        if data is None:
            self._reject()
            return None
        th = self._app_auth(data.get("token"))
        if th is None:
            self._reject()
            return None
        return th, self._acct_email_for_th(th), data

    def _agent_error(self, reason: str):
        """Map an agent refusal to (json body, status code) - mirrors
        the av_enqueue error mapping (409 quota, 404 missing, ...)."""
        code = {"no-ytdlp": 501, "no-ffmpeg": 501, "rate": 429,
                "full": 409, "size": 413, "bad": 400,
                "missing": 404}.get(reason, 400)
        return self._json({"ok": False, "error": reason}, code)

    def _h_agent_status(self):
        got = self._agent_auth()
        if got is None:
            return
        th, email, _data = got
        st = self.agent.status() if self.agent else {"ok": False}
        st["linked"] = bool(email)
        if email and self.agent:
            st["cookies"] = self.agent.cookies_status(email)
        return self._json({"ok": True, "agent": st})

    def _h_agent_dl_enqueue(self):
        got = self._agent_auth()
        if got is None:
            return
        th, email, data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        url = str(data.get("url", ""))
        mode = "video" if str(data.get("mode", "")) == "video" else "audio"
        if not url.startswith(("http://", "https://")):
            return self._json({"ok": False, "error": "url"}, 400)
        result = self.agent.dl_enqueue(email, th, url, mode,
                                       name=str(data.get("name", "")))
        if result == "ok":
            return self._json({"ok": True})
        return self._agent_error(result)

    def _h_agent_dl_list(self):
        got = self._agent_auth()
        if got is None:
            return
        _th, email, _data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        rows = self.agent.dl_list(email)
        now = time.time()
        for r in rows:
            r["age_s"] = max(0, int(now - float(r.get("updated", 0))))
        return self._json({"ok": True, "jobs": rows,
                           "active": self.agent.status().get("active", "")})

    def _h_agent_dl_clear(self):
        got = self._agent_auth()
        if got is None:
            return
        _th, email, _data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        self.agent.dl_clear(email)
        return self._json({"ok": True})

    def _h_agent_rec_add(self):
        got = self._agent_auth()
        if got is None:
            return
        th, email, data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        result = self.agent.rec_add(
            email, th, str(data.get("url", "")),
            str(data.get("name", "")), str(data.get("days", "")),
            str(data.get("time", "")), data.get("minutes", 30),
            tz_offset_min=data.get("tz_offset_min", 0))
        if result == "ok":
            return self._json({"ok": True})
        return self._agent_error(result)

    def _h_agent_rec_list(self):
        got = self._agent_auth()
        if got is None:
            return
        _th, email, _data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        return self._json({"ok": True,
                           "recs": self.agent.rec_list(email)})

    def _h_agent_rec_del(self):
        got = self._agent_auth()
        if got is None:
            return
        _th, email, data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        ok = self.agent.rec_del(email, str(data.get("id", "")))
        return self._json({"ok": ok}) if ok \
            else self._json({"ok": False, "error": "missing"}, 404)

    def _h_agent_cookies_put(self):
        # A real cookies.txt export exceeds the generic 4 KB body cap
        # (the 2026-10-02 'not found' bug), so this one route accepts
        # COOKIES_BODY_MAX; the worker still enforces its own store cap.
        got = self._agent_auth(cap=COOKIES_BODY_MAX, size_error=True)
        if got is None:
            return
        _th, email, data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        result = self.agent.cookies_put(email, str(data.get("cookies", "")))
        if result == "ok":
            return self._json({"ok": True})
        return self._agent_error(result)

    def _h_agent_cookies_status(self):
        got = self._agent_auth()
        if got is None:
            return
        _th, email, _data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        return self._json({"ok": True,
                           "cookies": self.agent.cookies_status(email)})

    def _h_agent_cookies_delete(self):
        got = self._agent_auth()
        if got is None:
            return
        _th, email, _data = got
        if not email:
            return self._json({"ok": False, "error": "no-account"}, 403)
        self.agent.cookies_delete(email)
        return self._json({"ok": True})

    def _h_phone_lib_list(self, code):
        """Library listing for a remembered device (owner-approved).
        Auth: device id must already be registered under the code."""
        data = self._body()
        if data is None:
            return self._reject()
        # Account session (phone page under /account/remote/<th>).
        th = self._ensure_account_pc(code)
        if th:
            rows = [{"id": l["id"], "name": l["name"],
                     "mb": round(l["bytes"] / 1e6, 1), "added": l["added"]}
                    for l in self.store.lib_list(th)]
            return self._json({"ok": True, "files": rows})
        devid = str(data.get("device_id", ""))[:64]
        if not devid or not self._pc_ref_ok(code):
            return self._reject()
        if not self.store.device_ok(code, devid):
            return self._reject()
        rows = [{"id": l["id"], "name": l["name"],
                 "mb": round(l["bytes"] / 1e6, 1), "added": l["added"]}
                for l in self.store.lib_list(self.store.th_for(code))]
        self._json({"ok": True, "files": rows})

    def _h_phone_lib_del(self, code):
        data = self._body()
        if data is None:
            return self._reject()
        # Account session (phone page under /account/remote/<th>).
        th = self._ensure_account_pc(code)
        if th:
            lid = str(data.get("id", ""))
            if not self.LIB_RE.match(lid):
                return self._reject()
            return self._json({"ok": self.store.lib_del(th, lid)})
        devid = str(data.get("device_id", ""))[:64]
        lid = str(data.get("id", ""))
        if not devid or not self.LIB_RE.match(lid) \
                or not self._pc_ref_ok(code):
            return self._reject()
        if not self.store.device_ok(code, devid):
            return self._reject()
        if not self.store.lib_del(self.store.th_for(code), lid):
            return self._json({"ok": False, "error": "missing"}, 404)
        self._json({"ok": True})

    def _send_audio(self, l):
        """Stream a library record with HTTP Range support (shared by the
        phone device path and the dashboard session path)."""
        size = l["bytes"]
        rng = self.headers.get("Range", "").strip()
        start, end = 0, size - 1
        status = 200
        m = re.match(r"bytes=(\d*)-(\d*)$", rng)
        if rng and not m:
            return self._json({"ok": False, "error": "range"}, 416)
        if m and m.group(1):
            start = int(m.group(1))
            if m.group(2):
                end = min(int(m.group(2)), size - 1)
            status = 206
        if start >= size or end < start:
            return self._json({"ok": False, "error": "range"}, 416)
        self.send_response(status)
        self.send_header("Content-Type", self._audio_type(l["name"]))
        self.send_header("Content-Length", str(end - start + 1))
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("Cache-Control", "private, no-store")
        if status == 206:
            self.send_header("Content-Range",
                             "bytes %d-%d/%d" % (start, end, size))
        self.end_headers()
        try:
            with open(l["file"], "rb") as f:
                f.seek(start)
                left = end - start + 1
                while left > 0:
                    chunk = f.read(min(65536, left))
                    if not chunk:
                        break
                    self.wfile.write(chunk)
                    left -= len(chunk)
        except (ConnectionError, BrokenPipeError):
            pass  # listener moved on (Range probe) - not an error
        except Exception:
            pass

    def _h_lib_audio(self, code, lid):
        """Stream a library file (Range-capable).  Auth: a remembered
        device (X-Device-Id header, the code phone page), the OWNER's
        account session for a linked PC (dashboard browser <audio> and
        the session phone page - browsers cannot set headers), in code
        or token-hash form."""
        if not self.cmd_rate.allow(self._ip()):
            return self._throttled()
        if not self._pc_ref_ok(code):
            return self._reject()
        th = None
        devid = str(self.headers.get("X-Device-Id", ""))[:64]
        if devid:
            # Read-only device check (by code or TH): never registers
            # anything - a probing stranger must not grow the table.
            if self.store.device_ok(code, devid):
                th = self.store.th_for(code)
        if th is None and self.accounts is not None:
            user = self._acct_user()
            if user is not None:
                th = self._ensure_account_pc(code)
        if th is None:
            return self._reject()
        l = self.store.lib_get(th, lid)
        if l is None or not os.path.isfile(l["file"]):
            return self._json({"ok": False, "error": "missing"}, 404)
        return self._send_audio(l)

    def _h_acct_lib_audio(self, lid):
        """Stream a stored library file to the OWNER's browser (dashboard,
        session cookie authed - browser <audio> cannot set headers).
        Resolves the file across every linked PC; Range-capable."""
        if not self.dash_rate.allow(self._ip()):
            return self._throttled()
        if not self.LIB_RE.match(lid):
            return self._reject()
        user = self._acct_user()
        if user is None:
            return self._reject()
        l = None
        for th, _label in self._linked_pcs(user["email"]):
            l = self.store.lib_get(th, lid)
            if l is not None:
                break
        if l is None or not os.path.isfile(l["file"]):
            return self._json({"ok": False, "error": "missing"}, 404)
        return self._send_audio(l)

    def _h_landing(self):
        body = self._PHONE_LANDING.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    # -- Listen Together rooms --------------------------------------------
    LT_STATE_KEYS = {"title", "url", "position", "playing", "live",
                     "ts"}

    def _h_lt_list(self):
        if not self.lt_rate.allow(self._ip()):
            return self._throttled()
        self._json({"ok": True, "rooms": self.rooms.list_public()})

    @staticmethod
    def _nick_body(data):
        """Room bodies accept only nick and name - nothing executable."""
        nick = str(data.get("nick", "")).strip()[:24]
        name = str(data.get("name", "")).strip()[:40]
        return nick, name

    def _h_lt_create(self):
        if not self.join_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        nick, name = self._nick_body(data)
        got = self.rooms.create(name, nick, self._ip())
        if got is None:
            return self._json({"ok": False, "error": "room limit"}, 503)
        rid, mid = got
        self._json({"ok": True, "room": rid, "member": mid})

    def _h_lt_join(self):
        if not self.join_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        nick, _ = self._nick_body(data)
        rid = str(data.get("room", ""))[:24]
        got = self.rooms.join(rid, nick, self._ip())
        if got == "gone":
            return self._json({"ok": False, "error": "room-gone"}, 404)
        if got == "banned":
            return self._json({"ok": False, "error": "banned"}, 403)
        if got == "full":
            return self._json({"ok": False, "error": "room-full"}, 403)
        rmid, mid, rname = got
        self._json({"ok": True, "room": rmid, "member": mid,
                    "name": rname})

    def _h_lt_poll(self):
        if not self.lt_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        got = self.rooms.poll(str(data.get("room", ""))[:24],
                              str(data.get("member", ""))[:24])
        if got is None:
            return self._json({"ok": False, "error": "not-in-room"}, 404)
        self._json({"ok": True, **got})

    def _h_lt_publish(self):
        if not self.lt_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        rid = str(data.get("room", ""))[:24]
        mid = str(data.get("member", ""))[:24]
        raw = data.get("state")
        state = {}
        if isinstance(raw, dict):
            for k in self.LT_STATE_KEYS & set(raw):
                v = raw[k]
                if k in ("title", "url"):
                    state[k] = str(v)[:500]
                elif k == "playing":
                    state[k] = bool(v)
                elif k == "live":
                    state[k] = bool(v)
                elif k == "position" or k == "ts":
                    try:
                        state[k] = max(0.0, float(v))
                    except (TypeError, ValueError):
                        pass
        if not self.rooms.publish(rid, mid, state):
            return self._json({"ok": False, "error": "not-in-room"}, 404)
        self._json({"ok": True})

    def _h_lt_leave(self):
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        self.rooms.leave(str(data.get("room", ""))[:24],
                         str(data.get("member", ""))[:24])
        self._json({"ok": True})

    def _h_lt_kick(self):
        """Room moderation: the HOST or a promoted ADMIN removes a member.

        The relay admin key still works (X-Admin-Key) and skips the
        permission ladder - it is the owner's override."""
        if not self.lt_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        forced = self._admin_ok()
        res = self.rooms.kick(str(data.get("room", ""))[:24],
                              str(data.get("actor", ""))[:24],
                              str(data.get("member", ""))[:24],
                              bool(data.get("ban")), force=forced)
        if res == "gone":
            return self._json({"ok": False, "error": "room-gone"}, 404)
        if res == "missing":
            return self._json({"ok": False, "error": "missing"}, 404)
        if res == "forbidden":
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True})

    def _h_lt_promote(self):
        """Host-only: grant/revoke moderator rights on a member."""
        if not self.lt_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        res = self.rooms.promote(str(data.get("room", ""))[:24],
                                 str(data.get("actor", ""))[:24],
                                 str(data.get("member", ""))[:24],
                                 bool(data.get("grant", True)))
        if res == "gone":
            return self._json({"ok": False, "error": "room-gone"}, 404)
        if res == "missing":
            return self._json({"ok": False, "error": "missing"}, 404)
        if res == "forbidden":
            return self._json({"ok": False, "error": "forbidden"}, 403)
        self._json({"ok": True})

    def _h_lt_voice_post(self):
        if not self.voice_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        try:
            rate = int(data.get("rate", 0))
        except (TypeError, ValueError):
            rate = 0
        res = self.rooms.voice_post(
            str(data.get("room", ""))[:24],
            str(data.get("member", ""))[:24],
            rate, str(data.get("pcm", "")))
        if res != "ok":
            return self._json({"ok": False, "error": res},
                              404 if res == "gone" else 400)
        self._json({"ok": True})

    def _h_lt_voice_sync(self):
        if not self.lt_rate.allow(self._ip()):
            return self._throttled()
        data = self._body()
        if data is None:
            return self._json({"ok": False, "error": "bad request"}, 400)
        try:
            since = max(0, int(data.get("since", 0)))
        except (TypeError, ValueError):
            since = 0
        got = self.rooms.voice_sync(str(data.get("room", ""))[:24],
                                    str(data.get("member", ""))[:24],
                                    since)
        if got is None:
            return self._json({"ok": False, "error": "not-in-room"}, 404)
        clips, seq = got
        self._json({"ok": True, "clips": clips, "seq": seq})

    def do_GET(self):
        path = self.path.split("?", 1)[0].rstrip("/") or "/"
        if path == "/app/lib/stream":
            # Token-authed range stream for the app itself (account
            # bookmark resume of a cloud track on another PC).
            return self._h_lib_stream()
        if path == "/healthz":
            return self._json({"ok": True})
        if path == "/":
            return self._h_landing()
        if path == "/admin":
            return self._h_admin_page()
        if path == "/admin/api/list":
            if not self.admin_rate.allow(self._ip()):
                return self._throttled()
            return self._h_admin_list()
        if path == "/admin/api/abuse":
            if not self.admin_rate.allow(self._ip()):
                return self._throttled()
            return self._h_admin_abuse()
        if path == "/lt/rooms":
            return self._h_lt_list()
        m = re.match(r"^/c/" + PCREF_RE + r"/state$", path)
        if m:
            return self._h_state(m.group(1))
        # Phones use the OWNER's account now (2026-09-25): a logged-in
        # session gets the phone page under /account/remote/<th> - also
        # honoured on the legacy /c/<code> URL.  Without a session the
        # classic code page still answers (old paired phones, bookmarks).
        if self.accounts is not None and self._acct_user() is not None:
            m = re.match(r"^/c/" + PCREF_RE + r"$", path)
            if m:
                return self._h_acct_remote_page(m.group(1))
        m = re.match(r"^/c/([A-Z2-9]{8,10})$", path)
        if m:
            return self._h_phone_page(m.group(1))
        # Live Room: the listener page (public, read-only, token in URL).
        m = re.match(r"^/lr/room/([0-9a-f]{64})$", path)
        if m:
            return self._h_lr_room_page(m.group(1))
        # Guest jukebox request page (public, share token in URL).
        m = re.match(r"^/juke/([A-Za-z0-9_-]{20,60})$", path)
        if m:
            return self._h_juke_guest_page(m.group(1))
        m = re.match(r"^/lr/room/([0-9a-f]{64})/stream$", path)
        if m:
            if not self.lr_rate.allow(self._ip()):
                return self._throttled()
            return self._h_lr_stream(m.group(1))
        m = re.match(r"^/lr/room/([0-9a-f]{64})/meta$", path)
        if m:
            if not self.lr_rate.allow(self._ip()):
                return self._throttled()
            return self._h_lr_meta(m.group(1))

        m = re.match(r"^/c/" + PCREF_RE + r"/lib/([A-Za-z0-9_-]{8,32})$", path)
        if m:
            return self._h_lib_audio(m.group(1), m.group(2))
        # Account pages (page_rate) and dashboard (acct_rate)
        if self.accounts is not None:
            # User feedback form (5.1): public, no account needed.
            if path == "/feedback":
                if not self.page_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_feedback_page()
            if path == "/account":
                # Phones land here now (2026-09-25): with a session, go
                # straight to the remote page (first linked PC); the
                # dashboard link is on every page.  Logged out -> login.
                if self._acct_user() is not None:
                    linked = self._linked_pcs(self._acct_user()["email"])
                    if linked:
                        return self._h_acct_redirect(
                            "remote/%s" % linked[0][0])
                    return self._h_acct_redirect("dashboard")
                return self._h_acct_redirect("login")
            if path in ("/account/signup", "/account/login",
                        "/account/forgot"):
                if not self.page_rate.allow(self._ip()):
                    return self._throttled()
                if path == "/account/signup":
                    return self._h_acct_signup()
                if path == "/account/login":
                    return self._h_acct_login_page()
                return self._h_acct_forgot_page()
            if path == "/account/verify":
                if not self.page_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_verify(
                    self.path.split("?", 1)[1] if "?" in self.path else "")
            if path == "/account/reset":
                if not self.page_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_reset_page(
                    self.path.split("?", 1)[1] if "?" in self.path else "")
            if path == "/account/dashboard":
                if not self.page_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_dashboard()
            # Session phone page + its fetch endpoints (cookie authed).
            m = re.match(r"^/account/remote/([A-Za-z0-9_-]{8,64})$", path)
            if m:
                if not self.page_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_remote_page(m.group(1))
            # browser <audio> streams the library by session (GET)
            m = re.match(r"^/account/remote/([A-Za-z0-9_-]{8,64})/"
                         r"lib/([A-Za-z0-9_-]{8,32})$", path)
            if m:
                if not self.cmd_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_lib_audio(m.group(1), m.group(2))
            m = re.match(r"^/account/lib/([A-Za-z0-9_-]{8,32})$", path)
            if m:
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_lib_audio(m.group(1))
        if path.startswith("/c/"):
            # malformed page path: same friendly miss page (no oracle)
            return self._h_phone_miss()
        self._reject()

    def do_POST(self):
        path = self.path.split("?", 1)[0].rstrip("/")
        # Live Room host endpoints (persistent-key authed, byte-pipe only).
        if path == "/lr/create":
            if not self.lr_rate.allow(self._ip()):
                return self._throttled()
            return self._h_lr_create()
        if path == "/lr/push":
            return self._h_lr_push()
        if path == "/lr/stop":
            return self._h_lr_stop()
        if path == "/lr/now":
            return self._h_lr_now()
        if path == "/lr/stats":
            return self._h_lr_stats()
        if path == "/pair":
            return self._h_pair()
        if path == "/app/sync":
            return self._h_sync()
        if path == "/revoke":
            return self._h_revoke()
        if path == "/app/device/revoke":
            return self._h_device_revoke()
        if path == "/app/lib/list":
            if not self.cmd_rate.allow(self._ip()):
                return self._throttled()
            return self._h_lib_list()
        if path == "/app/lib/del":
            if not self.cmd_rate.allow(self._ip()):
                return self._throttled()
            return self._h_lib_del()
        if path == "/app/lib/upload":
            return self._h_lib_upload()
        if path == "/app/av/enqueue":
            return self._h_av_enqueue()
        if path == "/app/av/status":
            return self._h_av_status()
        if path == "/app/agent/status":
            return self._h_agent_status()
        if path == "/app/agent/dl/enqueue":
            return self._h_agent_dl_enqueue()
        if path == "/app/agent/dl/list":
            return self._h_agent_dl_list()
        if path == "/app/agent/dl/clear":
            return self._h_agent_dl_clear()
        if path == "/app/agent/rec/add":
            return self._h_agent_rec_add()
        if path == "/app/agent/rec/list":
            return self._h_agent_rec_list()
        if path == "/app/agent/rec/del":
            return self._h_agent_rec_del()
        if path == "/app/agent/cookies/put":
            return self._h_agent_cookies_put()
        if path == "/app/agent/cookies/status":
            return self._h_agent_cookies_status()
        if path == "/app/agent/cookies/delete":
            return self._h_agent_cookies_delete()
        m = re.match(r"^/c/" + PCREF_RE + r"/liblist$", path)
        if m:
            return self._h_phone_lib_list(m.group(1))
        m = re.match(r"^/c/" + PCREF_RE + r"/libdel$", path)
        if m:
            return self._h_phone_lib_del(m.group(1))
        # Session phone page fetches (cookie authed; <id> = code or TH).
        m = re.match(r"^/account/remote/([A-Za-z0-9_-]{8,64})/"
                     r"(state|cmd|liblist|libdel)$", path)
        if m:
            ident, action = m.group(1), m.group(2)
            if action == "state":
                return self._h_state(ident)
            if action == "cmd":
                return self._h_cmd(ident)
            if action == "liblist":
                return self._h_phone_lib_list(ident)
            return self._h_phone_lib_del(ident)
        if path == "/admin/api/revoke":
            if not self.admin_rate.allow(self._ip()):
                return self._throttled()
            return self._h_admin_revoke()
        # (The old /admin/api/prank endpoint was removed entirely on
        # 2026-09-27 - unknown admin routes get the generic reject.)
        if path == "/lt/rooms":
            return self._h_lt_list()
        if path == "/lt/create":
            return self._h_lt_create()
        if path == "/lt/join":
            return self._h_lt_join()
        if path == "/lt/poll":
            return self._h_lt_poll()
        if path == "/lt/publish":
            return self._h_lt_publish()
        if path == "/lt/leave":
            return self._h_lt_leave()
        if path == "/lt/voice/post":
            return self._h_lt_voice_post()
        if path == "/lt/voice/sync":
            return self._h_lt_voice_sync()
        if path == "/lt/kick":
            return self._h_lt_kick()
        if path == "/lt/promote":
            return self._h_lt_promote()
        m = re.match(r"^/c/" + PCREF_RE + r"/cmd$", path)
        if m:
            return self._h_cmd(m.group(1))
        # Account endpoints (JSON + HTML forms + dashboard actions)
        if self.accounts is not None:
            # User feedback (5.1): JSON for the app, form post for the
            # site page.  The relay hosts the endpoint; the site form
            # posts here cross-origin (plain form post, no CORS need).
            if path in ("/feedback", "/feedback/api"):
                if not self.fb_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_feedback_submit()
            if path == "/account/api/register":
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_account_register()
            if path == "/account/api/roles":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_roles()
            if path == "/account/api/role/set":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_role_set()
            if path == "/account/api/feedback":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_feedback_list()
            if path == "/account/api/feedback/get":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_feedback_get()
            if path in ("/account/login", "/account/api/login"):
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                return self._handle_account_form("login")
            if path in ("/account/signup", "/account/api/signup"):
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                return self._handle_account_form("signup")
            if path in ("/account/forgot", "/account/api/forgot"):
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                return self._handle_account_form("forgot")
            if path == "/account/reset":
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                return self._handle_account_form("reset")
            if path == "/account/logout":
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                tok = self._session_from_cookie()
                if tok:
                    self.accounts.logout(tok)
                self.send_response(303)
                self._clear_session_cookie()
                self.send_header("Location", "login")
                self.send_header("Content-Length", "0")
                self.end_headers()
                return
            if path == "/account/link-code":
                if not self.acct_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_linkcode()
            if path == "/account/status":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_status()
            if path == "/account/pc/rename":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_pc_rename()
            if path == "/account/cmd":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_dash_cmd()
            if path == "/account/lib/list":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_lib_list()
            if path == "/account/lib/del":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_lib_del()
            if path == "/account/av/history":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_av_history()
            if path == "/account/device/rename":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_device_rename()
            if path == "/account/device/ban":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_acct_device_ban()
            if path == "/account/juke/state":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_juke_state()
            if path == "/account/juke/toggle":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_juke_toggle()
            if path == "/account/juke/decide":
                if not self.dash_rate.allow(self._ip()):
                    return self._throttled()
                return self._h_juke_decide()
            m = re.match(r"^/juke/([A-Za-z0-9_-]{20,60})/(list|req)$", path)
            if m:
                if not self.cmd_rate.allow(self._ip()):
                    return self._throttled()
                if m.group(2) == "list":
                    return self._h_juke_guest_list(m.group(1))
                return self._h_juke_guest_req(m.group(1))
            if path == "/app/account/link":
                return self._h_account_link()
            if path == "/app/account/unlink":
                return self._h_account_unlink()
            if path == "/app/bookmark/put":
                return self._h_bm_put()
            if path == "/app/bookmark/list":
                return self._h_bm_list()
            if path == "/app/bookmark/del":
                return self._h_bm_del()
            if path == "/app/position/put":
                return self._h_pos_put()
            if path == "/app/position/list":
                return self._h_pos_list()
            if path == "/app/stats/note":
                return self._h_stats_note()
            if path == "/app/settings/get":
                return self._h_settings_get()
            if path == "/app/settings/put":
                return self._h_settings_put()

        self._reject()


def reset_store_for_tests():
    """Fresh in-memory store + rate limiters (tests and manual runs).

    persist=False: tests must never read or write the real disk db."""
    _Handler.store = _Store(persist=False)
    _Handler.rooms = _Rooms()
    _Handler.pair_rate = _Rate(per_minute=6, burst=12)
    _Handler.sync_rate = _Rate(per_minute=120, burst=180)
    _Handler.cmd_rate = _Rate(per_minute=60, burst=120)
    _Handler.admin_rate = _Rate(per_minute=30, burst=60)
    _Handler.lt_rate = _Rate(per_minute=240, burst=360)
    _Handler.join_rate = _Rate(per_minute=12, burst=24)
    _Handler.voice_rate = _Rate(per_minute=40, burst=80)
    _Handler.lib_rate = _Rate(per_minute=90, burst=220)   # folder uploads OK
    _Handler.av_rate = _Rate(per_minute=20, burst=40)     # AudioVault jobs
    _Handler.av_worker = _AudioVaultWorker(_Handler.store)
    _Handler.abuse = _Abuse()
    _Handler.accounts = _accounts.Accounts(persist=False)
    # VPS agent AFTER accounts so job notifications can be queued; tests
    # get an unlimited limiter (the suite fires many requests per second).
    _Handler.agent = _vpsagent.ensure_worker(_Handler.store)
    _Handler.agent.set_notify(
        lambda email, kind, title, detail: _Handler.accounts.notify_user(
            email, kind, title, detail))
    _Handler.agent_rate = _Rate(per_minute=10 ** 6, burst=10 ** 6)
    _Handler.mailer = _accounts.Mailer()
    _Handler.acct_rate = _Rate(per_minute=10, burst=20)   # login/register
    _Handler.page_rate = _Rate(per_minute=60, burst=120)  # page GETs
    _Handler.dash_rate = _Rate(per_minute=60, burst=120)  # dashboard cmds
    _Handler.fb_rate = _Rate(per_minute=5, burst=10)      # feedback posts
    _Handler.start_alert_sweeper()
    _Handler.start_digest_sweeper()


def serve(host: str = "127.0.0.1", port: int = 8632) -> ThreadingHTTPServer:
    store = _Store()
    rooms = _Rooms()
    _Handler.store = store
    _Handler.rooms = rooms
    _Handler.pair_rate = _Rate(per_minute=6, burst=12)
    _Handler.sync_rate = _Rate(per_minute=120, burst=180)
    _Handler.cmd_rate = _Rate(per_minute=60, burst=120)
    _Handler.admin_rate = _Rate(per_minute=30, burst=60)
    _Handler.lt_rate = _Rate(per_minute=240, burst=360)
    _Handler.join_rate = _Rate(per_minute=12, burst=24)
    _Handler.voice_rate = _Rate(per_minute=40, burst=80)
    _Handler.lib_rate = _Rate(per_minute=90, burst=220)   # folder uploads OK
    _Handler.av_rate = _Rate(per_minute=20, burst=40)     # AudioVault jobs
    _Handler.av_worker = _AudioVaultWorker(store)
    _Handler.abuse = _Abuse()
    _Handler.accounts = _accounts.Accounts()      # env WAVELINE_ACCOUNTS_DB
    # VPS agent AFTER accounts (job notifications ride notify_user).
    _Handler.agent = _vpsagent.ensure_worker(store)
    _Handler.agent.set_notify(
        lambda email, kind, title, detail: _Handler.accounts.notify_user(
            email, kind, title, detail))
    _Handler.agent_rate = _Rate(per_minute=30, burst=60)  # agent jobs
    _Handler.mailer = _accounts.Mailer()          # env WAVELINE_SMTP_*
    _Handler.acct_rate = _Rate(per_minute=10, burst=20)
    _Handler.page_rate = _Rate(per_minute=60, burst=120)
    _Handler.dash_rate = _Rate(per_minute=60, burst=120)
    _Handler.fb_rate = _Rate(per_minute=5, burst=10)
    _Handler.start_alert_sweeper()
    _Handler.start_digest_sweeper()
    _Handler.admin_key = os.environ.get("RELAY_ADMIN_KEY", "")
    _Handler.lr = liveroom.LiveRooms()
    _Handler.lr_key = os.environ.get("WAVELINE_LR_KEY", "")
    _Handler.lr_rate = _Rate(per_minute=30, burst=60)

    def _sweeper():
        while True:
            time.sleep(600)
            try:
                store.sweep()
            except Exception:
                pass
            try:
                rooms.sweep()
            except Exception:
                pass
            try:
                _Handler.lr.sweep()
            except Exception:
                pass

    threading.Thread(target=_sweeper, daemon=True).start()
    httpd = ThreadingHTTPServer((host, port), _Handler)
    httpd.daemon_threads = True
    return httpd


if __name__ == "__main__":
    import os
    httpd = serve(port=int(os.environ.get("RELAY_PORT", "8632")))
    print("Waveline relay listening on %s:%d"
          % httpd.server_address[:2], flush=True)
    httpd.serve_forever()