"""
MeshBay Node — WebRTC DataChannel server for browser clients.

Browsers cannot use QUIC for NAT traversal (WebTransport doesn't allow choosing
the UDP source port — Port-Restricted Cone NAT requires exact port matching).
WebRTC DataChannel with ICE/STUN handles this automatically.

The MNP protocol (handshake, file_request, file_chunk, chat, etc.) runs
identically over WebRTC DataChannel as over QUIC streams. Same E2E encryption,
same message types, same msgpack wire format.

Wire format on the DataChannel:
  - Each message is length-prefixed msgpack (4-byte big-endian + msgpack payload)
  - Same as QUIC streams and TCP+TLS
  - DataChannel is ordered and reliable (SCTP over DTLS)

Signaling flow (handled externally by the hub):
  Browser → Hub : POST /v1/nodes/{id}/webrtc/offer  {sdp, ice_candidates}
  Hub → Node    : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id}
  Node → Hub    : WS push {type: "webrtc_answer", sdp, ice_candidates, peer_id}
  Hub → Browser : SSE/response {sdp, ice_candidates}
  After signaling, DataChannel is P2P — hub is out of the loop.
"""

import asyncio
import base64
import hashlib
import hmac
import logging
import os
import struct
import tempfile
import time
from pathlib import Path
from typing import Any

import blake3
import jwt
import msgpack
from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)

from meshbay_common import MNP_VERSION
from meshbay_common.handshake import (
    MNP_MIN_SUPPORTED,
    NONCE_LEN,
    ROLE_CLIENT,
    ROLE_NODE,
    HandshakeError,
    authorize_token,
    check_version,
    handshake_transcript,
    make_proof,
    verify_proof,
    webrtc_binding,
)
from meshbay_common.adminop import (
    ADMIN_CHALLENGE_TTL,
    OP_DIR_DELETE,
    OP_FILE_DELETE,
    OP_INVITE_CREATE,
    OP_MEMBER_REVOKE,
    OP_GEK_ROTATE,
    OP_MEMBER_UNPIN,
    OP_MEMBER_UPLOAD,
    OP_APPS_ENABLED,
    OP_SET_SCAN_SETTINGS,
    OP_TMDB_CONFIG,
    OP_TMDB_ENABLED,
    OP_VIDEO_ROOT,
    OP_TMDB_OVERRIDE,
    OP_TMDB_REMATCH,
    OP_MUSICBRAINZ_ENABLED,
    OP_AUDIO_ROOT,
    OP_PHOTO_ROOTS,
    OP_ROOT_ADD,
    OP_ROOT_REMOVE,
    OP_ROOT_UPDATE,
    OP_ROOT_EJECT,
    OP_ROOT_PLUG,
    OP_GROUP_ATTACH,
    OP_GROUP_DETACH,
    admin_transcript,
)
from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
from meshbay_common.device import (
    DEVICE_TTL,
    device_add_transcript,
    device_code_hash,
    device_request_transcript,
)
from meshbay_common.groupbox import PURPOSE_ACK, seal
from meshbay_common.join import (
    JOIN_TTL,
    ROLE_MEMBER,
    ROLE_OPERATOR,
    join_transcript,
)
from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire
from meshbay_node.transport.wire import index_sync_message
from meshbay_node.indexer import GroupIndex
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node import linkpreview, ops, platform
# Re-imported under its original name: every call site and existing test in
# this module still refers to it as `_probe_video`. The implementation lives
# in media_probe.py so the indexer package (imported just above) can call it
# too, for index-time enrichment, without a circular import.
from meshbay_node.media_probe import (
    BROWSER_INCOMPATIBLE_VIDEO_CODECS,
    probe_video as _probe_video,
)
from meshbay_node.roots import (
    RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name,
)

log = logging.getLogger(__name__)

CHUNK_SIZE = 1024 * 1024
MAX_MSG = 64 * 1024 * 1024

# Chat link-preview results, kept in memory only (draft-v6 §2.7: the node
# produces enrichment on demand and keeps nothing durable — the asking device
# caches). Bounded and time-limited so a busy group cannot grow it without end
# and a page that changed its card is picked up within the hour.
_LINK_PREVIEW_TTL = 3600
_LINK_PREVIEW_MAX = 256
_link_preview_cache: dict[str, tuple[float, dict]] = {}


def _link_preview_cache_get(url: str) -> dict | None:
    hit = _link_preview_cache.get(url)
    if hit is None:
        return None
    ts, value = hit
    if time.time() - ts > _LINK_PREVIEW_TTL:
        _link_preview_cache.pop(url, None)
        return None
    return value


def _link_preview_cache_put(url: str, value: dict) -> None:
    if not url:
        return
    if len(_link_preview_cache) >= _LINK_PREVIEW_MAX:
        oldest = min(_link_preview_cache, key=lambda k: _link_preview_cache[k][0])
        _link_preview_cache.pop(oldest, None)
    _link_preview_cache[url] = (time.time(), value)


# A member pasting a link is normal; a member — or a hub minting tokens for many
# accounts — firing hundreds is amplification/DoS and a way to make the node
# reach arbitrary hosts on demand (finding M3). Only a real outbound fetch is
# counted (a cache hit costs nothing), and the ceilings are generous enough that
# ordinary chat never meets them.
_LINK_PREVIEW_RATE_WINDOW = 60.0
_LINK_PREVIEW_RATE_PER_CONN = 15
_LINK_PREVIEW_RATE_NODE = 60

# Upload limits (finding C5a). Uploads used to land directly in the shared root under
# a name the client chose, overwriting whatever was already there — which both violated
# node sovereignty and defeated the delete authorization (overwrite a file, become its
# recorded uploader, then delete it legitimately).
MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file

# Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
# nowhere near enough to be a memory-exhaustion primitive (H6).
PRE_HANDSHAKE_MAX_MSG = 64 * 1024
# ffmpeg is spawned per stream request; without a cap any member can fork-bomb
# the node by requesting many streams at once (H6).
#
# Two was sized when a stream was a burst: the client took segments as fast as
# it could append them, so a slot was held for the minute it took to push the
# file and then came back. Now that the client only pulls ninety seconds ahead
# of the playhead, a slot is held for as long as the film runs — so two slots
# means two people can watch anything at all, and the third is refused for the
# next hour and a half. The work behind a slot has not changed and is small:
# ffmpeg runs `-c copy`, a remux with no encoding in it, and spends most of the
# film blocked on a pipe nobody is reading.
#
# This is the default, not the policy: the right number depends on the machine,
# so the operator sets `max_concurrent_streams` under [node] in node.toml. This
# value applies when they have said nothing.
MAX_CONCURRENT_TRANSCODES = 8
# Extensions no mainstream browser's <audio> element decodes natively, no
# matter how well-tagged (enrich_audio.py's problem) — the Music app's own
# analogue of BROWSER_INCOMPATIBLE_VIDEO_CODECS above, keyed by extension
# rather than a probed codec name since these two are a red flag on their
# own, not something that varies by how the file happens to be encoded
# inside.
BROWSER_INCOMPATIBLE_AUDIO_EXTS = frozenset({".wma", ".mpc"})
# A whole audio file is small enough to transcode in one shot rather than
# live-piped like video's fMP4 segments — a few seconds of ffmpeg at most,
# bounded generously so one slow/huge outlier can't pin a transcode slot
# (shared with video, MAX_CONCURRENT_TRANSCODES above) indefinitely.
AUDIO_TRANSCODE_TIMEOUT_SECS = 120
# Bundle fetches are served in the pre-proof window (C4). Bounded and audited
# until the native client removes remote keypair bundles entirely.
MAX_PRE_PROOF_FETCHES = 4
# Pairing codes carry 40 bits and are single-use, but a connection must not be
# allowed to sit there guessing. Failures are audited, so a grind is visible.
MAX_JOIN_ATTEMPTS = 5
# Per-connection limits alone would not bind an attacker who can open connections
# at will — and the adversary who can mint tokens for any account is the hub. So
# failed pairings are also counted node-wide over a window.
MAX_JOIN_FAILURES_WINDOW = 20
JOIN_FAILURE_WINDOW = 600  # seconds
# Everything a member sends lands here: files from the Files panel and
# attachments from the chat alike. One visible directory the operator can look
# into, back up or empty — rather than a hidden tree of per-user uuids that
# nobody could read, or files scattered wherever someone happened to be looking.
UPLOAD_DIR_NAME = "uploads"


def _extract_dtls_fingerprint(sdp: str) -> bytes:
    """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
    for line in sdp.splitlines():
        if line.startswith("a=fingerprint:sha-256 "):
            hex_str = line.split(" ", 1)[1].replace(":", "")
            return bytes.fromhex(hex_str)
    return b""


STREAM_SEGMENT_SIZE = 256 * 1024
# A chunk is a megabyte and the browser keeps eight in flight, so answering them
# as they arrive queues 8 MB on the channel with nothing watching. On a LAN that
# drains before anyone notices; on a phone that is also uploading, it is minutes
# of head-of-line delay for the reader. Above this, wait for room.
DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024
# What a client may ask for in one go, and how long the node waits for it to ask
# again before deciding nobody is watching any more.
STREAM_MAX_CREDIT = 256
STREAM_CREDIT_TIMEOUT = 120
# How often that budget is re-examined. A viewer who left stops being
# charged for a slot within this, rather than within the timeout.
STREAM_CREDIT_POLL = 3


def _pack(obj: dict) -> bytes:
    data = msgpack.packb(obj, use_bin_type=True)
    return struct.pack(">I", len(data)) + data


# Opt-in, off by default: a per-session heartbeat log (message count, time
# since the last message, ICE state) and ICE-state-change logging, on top of
# the connectionstatechange logging that already runs unconditionally. Added
# while chasing a report of the browser side going unresponsive after a
# mobile screen lock; --log-level DEBUG was not the right knob for this,
# since it is already used for the per-message request/response tracing
# every group index lookup produces, and turning that on for days of normal
# operation just to catch one intermittent session is not viable. Set
# MESHBAY_WEBRTC_TRACE=1 in the node's environment for the duration of a
# debugging session.
_WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1"
_WEBRTC_TRACE_INTERVAL_S = 30.0


class _DataChannelBuffer:
    """
    Accumulate DataChannel messages and extract length-prefixed msgpack.

    Finding H6: the limit was a flat 64 MB applied even before the handshake, so an
    unauthenticated peer could announce a 64 MB frame and dribble bytes into it,
    holding that much memory per connection. Until a peer has proved GEK
    possession it gets a small budget; the large one is for file uploads.
    """

    def __init__(self, max_message: int = MAX_MSG):
        self._buf = bytearray()
        self.max_message = max_message

    def feed(self, data: bytes):
        self._buf.extend(data)

    def messages(self):
        while len(self._buf) >= 4:
            length = struct.unpack(">I", self._buf[:4])[0]
            if length > self.max_message:
                raise ValueError(f"Message too large: {length}")
            if len(self._buf) < 4 + length:
                break
            msg_bytes = bytes(self._buf[4:4 + length])
            del self._buf[:4 + length]
            yield msgpack.unpackb(msg_bytes, raw=False)


def _get_remote_ip(pc: RTCPeerConnection) -> str:
    """Best-effort extraction of the remote peer IP from the ICE transport."""
    try:
        dtls = pc.sctp and pc.sctp.transport
        ice = dtls and dtls.transport
        conn = ice and ice._connection
        if conn and hasattr(conn, '_nominated') and conn._nominated:
            for pair in conn._nominated.values():
                return pair.remote_candidate.host
        if conn and conn.remote_candidates:
            return conn.remote_candidates[0].host
    except Exception:
        pass
    return ""


class WebRTCPeerSession:
    """One WebRTC peer connection, handling MNP over a DataChannel."""

    def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""):
        self._pc = pc
        self._ctx = node_ctx
        # Every background task this session starts. asyncio keeps only a *weak*
        # reference to a task, so one that is merely fired and forgotten can be
        # collected while it is still running — "Task was destroyed but it is
        # pending!" in the log. For _stream_video that meant its `async with
        # sem` never reached __aexit__ and the transcode slot was gone for good.
        # There are two slots: after two abandoned streams the node answered
        # "Server busy" to everything and no video would start at all.
        self._tasks: set[asyncio.Task] = set()
        self._channel: RTCDataChannel | None = None
        self._buffer = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
        self._pre_proof_fetches = 0
        self._user_id: str | None = None
        self._group_id: str | None = None
        self._peer_id: str = peer_id
        self._remote_ip: str = ""
        self._username: str = ""
        # Set from the roster: the key this node pinned for this account. Never
        # from the JWT — the hub picks what goes in there.
        self._pinned_pk: str = ""
        # Flow control for video: how many segments the client says it can take.
        self._stream_credit = 0
        self._stream_credit_evt = asyncio.Event()
        self._stream_stopped = False
        # When the peer last said anything about this stream. See
        # _await_stream_credit: silence is what ends a stream, not stinginess.
        self._stream_heard_at = 0.0
        # Diagnostics: how many `stream_more n=0` the peer sent. See
        # _grant_stream_credit — it tells a paced client from an unpaced one.
        self._stream_keepalives = 0
        # The stream this session currently owns. One viewer plays one film at
        # a time, so a second request means the first is over — see
        # _replace_stream for why waiting for it to time out is not an option.
        self._stream_task: asyncio.Task | None = None
        # Diagnostics only: when the current stream began and how far it got.
        self._stream_started_at: float = 0.0
        self._stream_segments: int = 0
        self._gek_challenge: bytes | None = None
        # Same value as the GEK challenge, but kept for the life of the connection:
        # a join_request is signed over it, and it must stay verifiable after the
        # handshake clears the challenge (an operator pairs while already connected).
        self._nonce_node: bytes = b""
        self._join_attempts = 0
        self._nonce_client: bytes = b""
        self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
        self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}
        # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message
        # arrived, so the heartbeat can report silence duration.
        self._last_msg_at: float = 0.0

    def _setup_channel(self, channel: RTCDataChannel) -> None:
        self._channel = channel
        self._msg_count = 0

        @channel.on("message")
        def on_message(message):
            if isinstance(message, str):
                message = message.encode()
            self._msg_count += 1
            self._last_msg_at = time.monotonic()
            if self._msg_count <= 3:
                log.info("WebRTC data received: %d bytes, msg #%d (peer=%s)",
                         len(message), self._msg_count, self._peer_id)
            self._buffer.feed(message)
            for msg in self._buffer.messages():
                self._handle_message(msg)

        if _WEBRTC_TRACE:
            self._spawn(self._trace_heartbeat())

    async def _trace_heartbeat(self) -> None:
        """Diagnostics only (_WEBRTC_TRACE): periodic proof-of-life for this
        session, so a gap in these lines pinpoints when the node stopped
        hearing from a peer that (from its own side) may still look connected."""
        while True:
            await asyncio.sleep(_WEBRTC_TRACE_INTERVAL_S)
            silence = time.monotonic() - self._last_msg_at if self._last_msg_at else -1
            log.info(
                "WebRTC heartbeat peer=%s msgs=%d silence=%.0fs pc=%s ice=%s",
                self._peer_id, self._msg_count, silence,
                self._pc.connectionState, self._pc.iceConnectionState,
            )

    def _handle_message(self, msg: dict) -> None:
        mtype = msg.get("type")
        log.debug("WebRTC recv: %s", mtype)
        try:
            if mtype == MNP.HANDSHAKE:
                self._do_handshake(msg)
            elif mtype == MNP.HANDSHAKE_RESPONSE:
                self._do_handshake_response(msg)
            elif mtype in (MNP.GEK_BUNDLE_FETCH, MNP.KEYPAIR_BUNDLE_FETCH) \
                    and self._gek_challenge is not None:
                # Served before the GEK proof by necessity: the client needs its
                # wrapped bundle in order to compute the proof. That window is a
                # disclosure surface (C4) — a hub that forges a JWT reaches it — so
                # it is bounded and audited here, and closed properly when clients
                # stop storing keypair bundles on other people's nodes.
                self._pre_proof_fetches += 1
                if self._pre_proof_fetches > MAX_PRE_PROOF_FETCHES:
                    self._audit_auth_failed(
                        getattr(self, "_pending_group", ""), "pre-proof fetch flood")
                    self._send({"type": "error", "detail": "Too many requests"})
                    return
                self._audit_pre_proof_fetch(mtype)
                if mtype == MNP.GEK_BUNDLE_FETCH:
                    self._spawn(self._do_gek_bundle_fetch())
                else:
                    self._spawn(self._do_keypair_bundle_fetch())
            elif mtype == MNP.JOIN_REQUEST and self._nonce_node:
                # Valid both before the GEK proof (a new member has no GEK to prove
                # with) and after it (an operator pairing a browser is already
                # connected). Authority comes from the pairing code and the
                # signature, never from the session state.
                self._spawn(self._do_join_request(msg))
            elif self._user_id is None:
                self._send({"type": "error", "detail": "Handshake required"})
            elif mtype == MNP.INDEX_SYNC:
                self._do_index_sync()
            elif mtype == MNP.FILE_REQUEST:
                # Spawned rather than answered inline: the reply waits for room
                # on the channel, and blocking the message loop for that would
                # stop everything else this peer is doing — including the
                # uploads whose acks free the very buffer we are waiting on.
                # Chunks are matched by file and index on the client, so
                # answering out of order is safe.
                self._spawn(self._do_file_request(msg))
            elif mtype == MNP.STREAM_SEGMENT:
                self._do_stream_segment(msg)
            elif mtype == MNP.CHAT_MESSAGE:
                self._do_chat_message(msg)
            elif mtype == MNP.CHAT_HISTORY:
                self._do_chat_history(msg)
            elif mtype == MNP.LINK_PREVIEW_REQ:
                self._spawn(self._do_link_preview_request(msg))
            elif mtype == MNP.PING:
                self._do_ping(msg)
            elif mtype == MNP.FILE_UPLOAD:
                self._do_file_upload(msg)
            elif mtype == MNP.DIR_CREATE:
                self._do_dir_create(msg)
            elif mtype == MNP.DIR_DELETE:
                self._do_dir_delete(msg)
            elif mtype == MNP.FILE_DELETE:
                self._do_file_delete(msg)
            elif mtype == MNP.ADMIN_RESPONSE:
                self._do_admin_response(msg)
            elif mtype == MNP.INVITE_CREATE:
                self._do_invite_create(msg)
            elif mtype == MNP.MEMBER_REVOKE:
                self._do_member_revoke(msg)
            elif mtype == MNP.DEVICE_REQUEST and self._nonce_node:
                self._spawn(self._do_device_request(msg))
            elif mtype == MNP.DEVICE_LOOKUP:
                self._spawn(self._do_device_lookup(msg))
            elif mtype == MNP.DEVICE_ADD:
                self._spawn(self._do_device_add(msg))
            elif mtype == MNP.DEVICE_LIST:
                self._spawn(self._do_device_list(msg))
            elif mtype == MNP.DEVICE_REVOKE:
                self._spawn(self._do_device_revoke(msg))
            elif mtype == MNP.MEMBER_UPLOAD:
                self._do_member_upload(msg)
            elif mtype == MNP.APPS_ENABLED:
                self._do_apps_enabled(msg)
            elif mtype == MNP.SET_SCAN_SETTINGS:
                self._do_set_scan_settings(msg)
            elif mtype == MNP.TMDB_CONFIG:
                self._do_tmdb_config(msg)
            elif mtype == MNP.TMDB_ENABLED:
                self._do_tmdb_enabled(msg)
            elif mtype == MNP.VIDEO_ROOT:
                self._do_video_root(msg)
            elif mtype == MNP.AUDIO_ROOT:
                self._do_audio_root(msg)
            elif mtype == MNP.PHOTO_ROOTS:
                self._do_photo_roots(msg)
            elif mtype == MNP.MEDIA_META_REQ:
                self._spawn(self._do_media_meta_request(msg))
            elif mtype == MNP.SEASON_META_REQ:
                self._spawn(self._do_season_meta_request(msg))
            elif mtype == MNP.TMDB_SEARCH_REQ:
                self._spawn(self._do_tmdb_search_request(msg))
            elif mtype == MNP.TMDB_OVERRIDE:
                self._do_tmdb_override(msg)
            elif mtype == MNP.TMDB_REMATCH:
                self._do_tmdb_rematch(msg)
            elif mtype == MNP.MUSICBRAINZ_ENABLED:
                self._do_musicbrainz_enabled(msg)
            elif mtype == MNP.MUSIC_META_REQ:
                self._spawn(self._do_music_meta_request(msg))
            elif mtype == MNP.AUDIO_TRANSCODE_REQ:
                self._spawn(self._do_audio_transcode_request(msg))
            elif mtype == MNP.MEMBER_UNPIN:
                self._do_member_unpin(msg)
            elif mtype == MNP.GEK_ROTATE:
                self._do_gek_rotate(msg)
            elif mtype == MNP.NODE_STATUS:
                self._spawn(self._do_node_status(msg))
            elif mtype == MNP.ROOT_ADD:
                self._do_root_add(msg)
            elif mtype == MNP.ROOT_REMOVE:
                self._do_root_remove(msg)
            elif mtype == MNP.ROOT_UPDATE:
                self._do_root_update(msg)
            elif mtype == MNP.ROOT_EJECT:
                self._do_root_eject(msg)
            elif mtype == MNP.ROOT_PLUG:
                self._do_root_plug(msg)
            elif mtype == MNP.ROSTER_READ:
                self._spawn(self._do_roster_read(msg))
            elif mtype == MNP.DENYLIST_READ:
                self._spawn(self._do_denylist_read(msg))
            elif mtype == MNP.DENYLIST_CLEAR:
                self._spawn(self._do_denylist_clear(msg))
            elif mtype == MNP.GROUP_ATTACH:
                self._do_group_attach(msg)
            elif mtype == MNP.GROUP_DETACH:
                self._do_group_detach(msg)
            elif mtype == MNP.NODE_SETTINGS_SET:
                self._spawn(self._do_node_settings_set(msg))
            elif mtype == MNP.NODE_RELOAD:
                self._spawn(self._do_node_reload(msg))
            elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
                self._spawn(self._do_keypair_bundle_store(msg))
            elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
                self._spawn(self._do_keypair_bundle_delete())
            elif mtype == MNP.STREAM_REQUEST:
                sem = self._ctx.get("_transcode_sem")
                log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
                         str(msg.get("file_id"))[:12], msg.get("credits"),
                         getattr(sem, "_value", "?"),
                         "alive" if (self._stream_task and
                                     not self._stream_task.done()) else "none")
                self._spawn(self._replace_stream(msg))
            elif mtype == MNP.STREAM_MORE:
                self._grant_stream_credit(msg)
            elif mtype == "client_diag":
                # Diagnostics only. The node acts on none of it — it writes it
                # next to its own view of the same stream, which is the only
                # place the two halves can be compared when the client is a
                # phone with no console.
                # Every field is peer-controlled, so each is stringified and
                # cut short: this is a log line, not a channel for writing
                # whatever one likes into the operator's file.
                def _f(key: str, n: int = 24) -> str:
                    return str(msg.get(key))[:n].replace("\n", " ")
                if msg.get("event"):
                    # Once per stream or per seek, not once per five seconds —
                    # and a seek nobody asked for looks exactly like a viewer
                    # dragging the scrubber from this side, so it has to be
                    # visible without turning DEBUG on.
                    log.info(
                        "stream: client %s target=%s t=%ss offset=%s ready=%s "
                        "duration=%s ranges=[%s]",
                        _f("event", 16), _f("target"), _f("t"), _f("offset"),
                        _f("ready"), _f("duration"), _f("ranges", 120))
                # Debug: one line every five seconds per viewer. Run the daemon
                # with --log-level debug to see inside a player that is
                # misbehaving — it is the only view of the browser there is
                # when the browser is a phone.
                else:
                    log.debug(
                        "stream: client t=%ss ahead=%ss ready=%s paused=%s "
                        "stalled=%s q=%s inflight=%s appending=%s updating=%s "
                        "quota=%s ms=%s err=%s ranges=[%s] (sent=%d)",
                        _f("t"), _f("ahead"), _f("ready"), _f("paused"),
                        _f("stalled"), _f("q"), _f("inflight"), _f("appending"),
                        _f("updating"), _f("quota"), _f("ms"), _f("err", 80),
                        _f("ranges", 120), self._stream_segments)
            elif mtype == MNP.STREAM_STOP:
                age = (time.monotonic() - self._stream_started_at
                       if self._stream_started_at else -1)
                log.info("stream: stop received %.1fs after start, %d segments sent",
                         age, self._stream_segments)
                self._stop_stream()
            else:
                log.warning("Unknown MNP message type on DataChannel: %s", mtype)
        except Exception as e:
            # Log the detail locally; send the peer a generic message. Exception
            # text here carries filesystem paths and internal state (finding L3).
            log.error("Error handling %s on DataChannel: %s", mtype, e, exc_info=True)
            self._send({"type": "error", "detail": "Request failed"})

    def _audit(self, event: str, detail: str = "") -> None:
        audit = self._ctx.get("audit_store")
        if audit and self._user_id:
            if not self._remote_ip:
                self._remote_ip = _get_remote_ip(self._pc)
            self._spawn(audit.log_event(
                user_id=self._user_id,
                event=event,
                ip=self._remote_ip,
                username=self._username,
                group_id=self._group_id or "",
                detail=detail,
            ))

    def _channel_binding(self) -> bytes:
        """Both DTLS fingerprints, so a proof is valid on this connection only."""
        offer_fp = b""
        answer_fp = b""
        if self._pc.remoteDescription:
            offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp)
        if self._pc.localDescription:
            answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp)
        if not offer_fp or not answer_fp:
            return b""
        return webrtc_binding(offer_fp, answer_fp)

    def _do_handshake(self, msg: dict) -> None:
        group_id = msg.get("group_id", "")
        log.info("WebRTC handshake request: group=%s (peer=%s)",
                 group_id[:8] if group_id else "none", self._peer_id)
        # Before the token, and before anything is decided from it: a peer we
        # cannot speak to is refused with a code it can act on, rather than
        # served messages it will misread as missing fields (L2).
        try:
            check_version(msg.get("v", ""), msg.get("v_min", ""))
        except HandshakeError as refusal:
            self._send({"type": "error", "detail": str(refusal),
                        "code": refusal.code})
            return
        try:
            peer = authorize_token(
                msg.get("token", ""),
                self._ctx["hub_pk_pem"],
                group_id=group_id,
                hosted_groups=self._ctx.get("groups"),
                denylist=self._ctx.get("denylist"),
            )
        except HandshakeError as refusal:
            # HandshakeError messages are authored to be peer-safe, unlike arbitrary
            # exception text (L3) — the client needs to know *why* it was refused.
            self._send({"type": "error", "detail": str(refusal),
                        "code": getattr(refusal, "code", "")})
            self._audit_auth_failed(group_id, str(refusal))
            return

        try:
            self._nonce_client = base64.b64decode(msg.get("nonce", ""))
        except Exception:
            self._nonce_client = b""
        if len(self._nonce_client) < NONCE_LEN:
            # The client nonce is what makes the NODE's proof fresh (C3). Without
            # it a recorded ack could be replayed by an impersonating peer.
            self._send({"type": "error", "detail": "Client nonce required"})
            return

        # Decoded, but NOT authenticated: that happens on the GEK proof.
        self._pending_sub = peer.user_id
        self._pending_group = peer.group_id
        self._pending_username = peer.username

        gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx
        if not gctx.get("gek"):
            log.warning("Handshake refused — no GEK for group=%s", peer.group_id[:8])
            self._send({
                "type": "error",
                "detail": "Group encryption not initialized — contact node operator",
            })
            return

        self._gek_challenge = os.urandom(NONCE_LEN)
        self._nonce_node = self._gek_challenge
        log.info("WebRTC handshake challenge sent (peer=%s)", self._peer_id)
        self._send({
            "type": MNP.HANDSHAKE_CHALLENGE,
            "v": MNP_VERSION,
            # Our half of the range. The client refuses us on this rather than
            # discovering the mismatch when a field it expected is not there.
            "v_min": MNP_MIN_SUPPORTED,
            "nonce": base64.b64encode(self._gek_challenge).decode(),
            # Announced here because a first-time joiner needs it *before* the
            # ack: join_request signs a transcript naming this node, and someone
            # who has never held the GEK cannot complete the handshake to learn
            # it. Unverified at this point — the ack proves it, the client checks
            # the two match, and a wrong value only makes our own verification
            # fail. It is never a substitute for the ack's proof and signature.
            "node_pk": self._node_pk_b64(),
        })

    def _do_handshake_response(self, msg: dict) -> None:
        if not self._gek_challenge or not hasattr(self, "_pending_sub"):
            self._send({"type": "error", "detail": "No pending handshake challenge"})
            return

        group_id = self._pending_group
        gctx = self._ctx["groups"][group_id] if "groups" in self._ctx else self._ctx
        gek = gctx.get("gek")
        if not gek:
            self._send({"type": "error", "detail": "Group encryption not initialized"})
            self._gek_challenge = None
            return

        try:
            proof_bytes = base64.b64decode(msg.get("proof", ""))
        except Exception:
            self._send({"type": "error", "detail": "Invalid proof encoding"})
            return

        binding = self._channel_binding()
        if not binding:
            # Refuse rather than fall back to an unbound proof (L4).
            self._send({"type": "error", "detail": "Channel binding unavailable"})
            self._gek_challenge = None
            self._audit_auth_failed(group_id, "no channel binding")
            return

        if not verify_proof(gek, proof_bytes, ROLE_CLIENT, group_id,
                            self._nonce_client, self._gek_challenge, binding):
            self._send({"type": "error", "detail": "GEK proof failed"})
            self._gek_challenge = None
            self._audit_auth_failed(group_id, "GEK HMAC mismatch")
            return

        self._complete_handshake(gek, binding)
        self._gek_challenge = None

    def _complete_handshake(self, gek: bytes, binding: bytes) -> None:
        # Authenticated peers may send large frames (file uploads); unauthenticated
        # ones may not (H6).
        self._buffer.max_message = MAX_MSG
        self._user_id = self._pending_sub
        self._group_id = self._pending_group
        self._username = self._pending_username
        self._spawn(self._load_pinned_pk())

        self._peer_registry()[self._user_id] = self

        node_user_id = self._ctx.get("node_user_id")
        log.info("WebRTC handshake OK — user=%s group=%s",
                 self._user_id[:8],
                 self._group_id[:8] if self._group_id else "none")
        # The node proves itself too (C3): possession of the GEK over the client's
        # nonce, plus a signature over the same transcript with its long-term key.
        # Previously the client received an unverifiable node_pk and trusted
        # is_node_admin from whoever answered — so a peer that had hijacked
        # signaling could serve a forged index, chat history and permissions.
        node_transcript = handshake_transcript(
            ROLE_NODE, self._group_id or "", self._nonce_client,
            self._gek_challenge or b"", binding)
        node_proof = make_proof(
            gek, ROLE_NODE, self._group_id or "", self._nonce_client,
            self._gek_challenge or b"", binding)

        # Everything the client needs in order to *authenticate* us stays in clear —
        # node_pk, proof and sig are what it checks before it would trust a
        # decryption, so they cannot themselves be behind one. The configuration
        # below is sealed under a GEK-derived subkey, which gives it an
        # authentication tag from a key the hub does not hold. Until MNP 1.0 the
        # signed transcript named no ack field at all, so is_node_admin,
        # enabled_apps, video_root and the rest were authenticated by the DTLS
        # channel and nothing else.
        config = {
            "is_node_admin": self._is_node_admin(),
            # Backward compat for MNP 1.0 clients: computed from writable roots.
            # New clients read per-root writable from the index payload instead.
            "member_upload": any(
                r.get("writable") for r in
                (self._group_ctx().get("roots").describe()
                 if self._group_ctx().get("roots") else [])),
            # Which group "applications" to show. Absent/empty falls back to
            # every registered one client-side, so a node that predates this
            # setting (or one whose context has not loaded it yet) hides
            # nothing.
            "enabled_apps": list(self._group_ctx().get("enabled_apps") or []),
            # Which folder the Videos app treats as its entry point for
            # this group — "" means the whole group index.
            "video_root": self._group_ctx().get("video_root") or "",
            # Per-group (2026-08-24 — used to be node-wide), same "read once,
            # kept current in place by the signed op" shape as video_root
            # above — surfaced here rather than only via tmdb_enabled_ack so
            # a client that connects after the operator already configured
            # it does not have to wait for a live change to find out.
            "tmdb_enabled": bool(self._group_ctx().get("tmdb_enabled", True)),
            # Token/language stay node-wide (one shared credential/cache) —
            # via daemon_state, kept current by tmdb_config_ack.
            "tmdb_token_customized": bool(
                self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)),
            "tmdb_language": str(
                self._ctx.get("daemon_state", {}).get("tmdb_language") or ""),
            # Music app (docs/musicbay.md §6) — same shape as the TMDB
            # fields above. No language field: MusicBrainz search doesn't
            # take one the way TMDB does.
            "musicbrainz_enabled": bool(self._group_ctx().get("musicbrainz_enabled", True)),
            # Which folder the Music app treats as its entry point for this
            # group — same shape as video_root above, "" means unset (the
            # Music tab shows nothing yet).
            "audio_root": self._group_ctx().get("audio_root") or "",
            # Which folder(s) the Photos app treats as its entry points for
            # this group — a *list*, unlike video_root/audio_root above
            # (docs/photos.md §2.1). Empty means unset (the Photos tab shows
            # nothing yet).
            "photo_roots": list(self._group_ctx().get("photo_roots") or []),
            # So a client that connects mid-scan shows the indexing state
            # immediately, instead of waiting for the next periodic
            # INDEX_PROGRESS push. Never a path or filename — see
            # IndexProgress in indexer.py.
            "indexing": self._indexing_status(),
            # Current values only — not enforced from here, just shown to
            # the operator in Settings so the number on screen matches what
            # the indexer is actually doing (set_scan_settings, ops.py).
            "scan_settings": {
                "reconcile_interval_secs": self._group_ctx().get(
                    "reconcile_interval_secs", DirectoryIndexer.DEFAULT_RECONCILE_SECS),
                "debounce_secs": self._group_ctx().get(
                    "debounce_secs", DirectoryIndexer.DEFAULT_DEBOUNCE_SECS),
            },
        }
        if node_user_id:
            config["node_user_id"] = node_user_id
        pk_x_b64 = self._ctx.get("pk_x25519_b64")
        if pk_x_b64:
            config["node_pk_x25519"] = pk_x_b64

        ack = {
            "type": MNP.HANDSHAKE_ACK,
            "v": MNP_VERSION,
            "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()),
            "proof": base64.b64encode(node_proof).decode(),
            "sig": base64.b64encode(
                self._ctx["sk_node"].sign(node_transcript)).decode(),
            **seal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, self._group_id or "", config),
        }
        self._send(ack)
        self._audit("handshake")

        # Someone is here now — reconcile's backstop should be prompt again
        # rather than however far its backoff had stretched while nobody
        # was connected (indexer.py DirectoryIndexer.note_activity).
        note_activity = self._group_ctx().get("note_activity")
        if note_activity:
            note_activity()

    async def _do_gek_bundle_fetch(self) -> None:
        """Serve the caller's wrapped GEK bundle during the handshake window."""
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
            return

        group_id = getattr(self, "_pending_group", "")
        user_id = getattr(self, "_pending_sub", "")
        if not group_id or not user_id:
            self._send({"type": "error", "detail": "No pending handshake"})
            return

        bundle = await bundle_store.fetch(group_id, user_id)
        if bundle:
            self._send({
                "type": MNP.GEK_BUNDLE_RESP,
                "v": MNP_VERSION,
                "found": True,
                "pk_eph_b64": bundle["pk_eph_b64"],
                "nonce_b64": bundle["nonce_b64"],
                "wrapped_b64": bundle["wrapped_b64"],
            })
        else:
            self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})

    def _do_invite_create(self, msg: dict) -> None:
        """
        Issue a one-time pairing code for someone the operator wants to admit.

        Replaces the old invite path, where the inviter fetched the invitee's
        public key from the hub and wrapped the group key for whatever came back
        (H3). The node now needs nothing but a name: it will wrap the key itself,
        later, for a key the invitee proves they hold.
        """
        roster = self._ctx.get("roster")
        if roster is None:
            self._send({"type": "error", "detail": "Roster not available"})
            return

        invitee_id = msg.get("user_id", "")
        group_id = msg.get("group_id") or self._group_id
        if not invitee_id or not group_id:
            self._send({"type": "error", "detail": "Missing user_id or group_id"})
            return
        if group_id != self._group_id:
            self._send({"type": "error", "detail": "Wrong group for this session"})
            return

        if not self._has_admin_authority():
            self._send({
                "type": "error",
                "detail": "No operator paired — run `meshbay-node operator pair`",
            })
            return

        self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, {
            "group_id": group_id,
            "user_id": invitee_id,
            "username": str(msg.get("username", ""))[:64],
        })

    async def _do_keypair_bundle_fetch(self) -> None:
        """Serve the caller's encrypted keypair bundle during the handshake window."""
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
            return

        user_id = getattr(self, "_pending_sub", "")
        if not user_id:
            self._send({"type": "error", "detail": "No pending handshake"})
            return

        kp = await bundle_store.fetch_keypair(user_id)
        if kp and kp.get("bundle_enc"):
            resp = {
                "type": MNP.KEYPAIR_BUNDLE_RESP,
                "v": MNP_VERSION,
                "found": True,
                "bundle_enc": kp["bundle_enc"],
            }
            # The recovery-wrapped copy (MNP 0.14) rides along when present, so a
            # client holding the recovery key can re-wrap it under a new
            # passphrase — docs/auth-confirm.md §4.5.
            if kp.get("bundle_enc_recovery"):
                resp["bundle_enc_recovery"] = kp["bundle_enc_recovery"]
            self._send(resp)
        else:
            self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False})

    async def _do_keypair_bundle_store(self, msg: dict) -> None:
        """Store an encrypted keypair bundle (user backs up their own keys on node)."""
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": "error", "detail": "Bundle store not available"})
            return

        bundle_enc = msg.get("bundle_enc", "")
        if not bundle_enc:
            self._send({"type": "error", "detail": "Missing bundle_enc"})
            return

        # Optional second copy wrapped under the recovery key (MNP 0.14). Omitted
        # by an older client and by a plain re-backup; the store keeps any
        # existing recovery copy when this is absent.
        recovery = msg.get("bundle_enc_recovery") or None

        await bundle_store.store_keypair(self._user_id, bundle_enc, recovery)
        log.info("Keypair bundle stored for user=%s (recovery=%s)",
                 self._user_id[:8], bool(recovery))
        self._audit("keypair_bundle_store")
        self._send({
            "type": "ack", "v": MNP_VERSION,
            "detail": "keypair_bundle_stored",
        })

    # ── Pairing and join (H3, M3) ────────────────────────────────────────────

    def _join_refuse(self, reason: str, audit_detail: str = "") -> None:
        self._join_attempts += 1
        # Node-wide window, shared across connections: reconnecting must not reset
        # the budget.
        now = time.time()
        failures = [t for t in self._ctx.get("join_failures", [])
                    if now - t < JOIN_FAILURE_WINDOW]
        failures.append(now)
        self._ctx["join_failures"] = failures
        self._audit_join("join_refused", audit_detail or reason)
        self._send({
            "type": MNP.JOIN_RESULT,
            "v": MNP_VERSION,
            "ok": False,
            "reason": reason,
        })

    def _audit_join(self, event: str, detail: str) -> None:
        audit = self._ctx.get("audit_store")
        if not audit:
            return
        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
        self._spawn(audit.log_event(
            user_id=self._user_id or getattr(self, "_pending_sub", "unknown"),
            event=event,
            ip=self._remote_ip,
            username=self._username or getattr(self, "_pending_username", ""),
            group_id=self._group_id or getattr(self, "_pending_group", "") or "",
            detail=detail,
        ))

    async def _do_join_request(self, msg: dict) -> None:
        """
        Pin an identity, or recognise one already pinned.

        The client signs its own Ed25519 and X25519 keys together with the node's
        nonce, so the identity key vouches for the encryption key — that is what
        will make it safe for the node to wrap the GEK for a key that arrived over
        the wire instead of one fetched from the hub's directory (H3).

        A first pairing needs a one-time code, which the hub never sees. Afterwards
        the pin is the credential and a changed key is refused outright, the same
        rule the client applies to `pk_node` (11.5.8).
        """
        roster = self._ctx.get("roster")
        if roster is None:
            self._send({"type": "error", "detail": "Roster not available"})
            return

        if self._join_attempts >= MAX_JOIN_ATTEMPTS:
            self._send({"type": "error", "detail": "Too many attempts"})
            return

        now = time.time()
        recent = [t for t in self._ctx.get("join_failures", [])
                  if now - t < JOIN_FAILURE_WINDOW]
        if len(recent) >= MAX_JOIN_FAILURES_WINDOW:
            self._audit_join("join_throttled", f"{len(recent)} failures in window")
            self._send({"type": "error", "detail": "Pairing temporarily locked"})
            return

        user_id = self._user_id or getattr(self, "_pending_sub", "")
        username = self._username or getattr(self, "_pending_username", "")
        if not user_id:
            self._send({"type": "error", "detail": "Handshake required"})
            return

        pk_ed_b64 = msg.get("pk_ed25519", "")
        pk_x_b64 = msg.get("pk_x25519", "")
        code = msg.get("code", "")
        ts = msg.get("ts", 0)

        try:
            pk_ed_raw = base64.b64decode(pk_ed_b64)
            pk_x_raw = base64.b64decode(pk_x_b64)
            if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32:
                raise ValueError
            pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw)
        except Exception:
            self._join_refuse("invalid_keys")
            return

        if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL:
            self._join_refuse("stale_request")
            return

        # An empty group_id means operator pairing, which is node-wide. Anything
        # else must be the group this connection authenticated to — a signature
        # obtained for one group must not name another.
        group_id = msg.get("group_id", "") or ""
        session_group = self._group_id or getattr(self, "_pending_group", "") or ""
        if group_id and group_id != session_group:
            self._join_refuse("group_mismatch")
            return

        transcript = join_transcript(
            node_pk_b64=self._node_pk_b64(),
            group_id=group_id,
            user_id=user_id,
            pk_ed25519_b64=pk_ed_b64,
            pk_x25519_b64=pk_x_b64,
            nonce_node=self._nonce_node,
            ts=ts,
        )
        try:
            sig = base64.b64decode(msg.get("sig", ""))
        except Exception:
            self._join_refuse("invalid_signature_encoding")
            return
        if not self._verify_sig(pk_ed, transcript, sig):
            self._join_refuse("signature_invalid")
            return

        # One person may hold several devices here — a browser and a desktop
        # client are two keys on one account. So the question is not "is this
        # THE key" but "is this ONE OF this account's live devices".
        device = await roster.find_device(user_id, pk_ed_b64)
        if device and device["pk_x25519"] != pk_x_b64:
            # The Ed25519 key is pinned but arrives with a different encryption
            # key. The join transcript signs both together, so this is either a
            # client that regenerated half its identity or something splicing
            # two messages; either way the pair is not the one admitted.
            self._join_refuse(
                "key_changed",
                f"pinned x25519={device['pk_x25519'][:16]} presented={pk_x_b64[:16]}")
            return
        known = device
        if not known and await roster.list_devices(user_id):
            # The account is known here but this key is not one of its devices.
            # Not an error to shout about: it is a second browser or a new
            # client, and the way in is a device-add approved by a device that
            # is already trusted — no operator, no new invitation code.
            self._join_refuse(
                "unknown_device",
                f"presented={pk_ed_b64[:16]} — approve it from a device already "
                f"paired with this node")
            return
        if known:
            # This group's own row first; then the join message's group_id (empty
            # on the node-wide first connect); then the operator's node-wide row,
            # which is where an operator opening any group finds their authority.
            member = (await roster.get_member(session_group, user_id)
                      or await roster.get_member(group_id, user_id)
                      or await roster.get_member("", user_id))
            if not member and self._group_join_policy(session_group) == "open":
                await roster.set_member(
                    group_id=session_group, user_id=user_id, role=ROLE_MEMBER,
                    status="active", approved_by="open-join",
                )
                member = await roster.get_member(session_group, user_id)

            # A pending invite means the operator explicitly re-invited this
            # person — require the code even if they already have a member
            # row (e.g. they left and were re-invited, or were revoked then
            # re-invited).  Without this gate a stale roster row lets them
            # back in without proving they received the new code.
            pending_invite = any(
                i["user_id"] == user_id
                and i["group_id"] in (session_group, "")
                for i in await roster.list_invites())
            if pending_invite:
                if not code:
                    self._join_refuse("code_required")
                    return
                invite = await roster.consume_invite(code, user_id)
                if not invite:
                    self._join_refuse("code_invalid")
                    return
                await roster.set_member(
                    group_id=invite["group_id"], user_id=user_id,
                    role=invite["role"], status="active",
                    approved_by=invite["created_by"],
                )
                self._audit_join(
                    "join_pinned",
                    f"group={invite['group_id'][:8]} role={invite['role']} "
                    "via=code (device already known)")
                member = (await roster.get_member(session_group, user_id)
                          or await roster.get_member(invite["group_id"], user_id))

            if not member:
                self._join_refuse("not_authorized_for_group")
                return
            await self._join_ok(
                user_id, pk_x_raw, session_group,
                role=member["role"] if member else "",
                recognised=True,
            )
            return

        if not code:
            if self._group_join_policy(session_group) == "open":
                # An open-join group admits anyone the hub calls a member, so a
                # code would protect nothing — the hub can walk in through the
                # front door. Pin what turns up and say so in the audit log.
                await self._pin_and_admit(
                    roster, user_id, username, pk_ed_b64, pk_x_b64,
                    group_id=session_group, role=ROLE_MEMBER,
                    approved_by="open-join", via="tofu")
                await self._join_ok(user_id, pk_x_raw, session_group,
                                    role=ROLE_MEMBER, recognised=False)
                return
            self._join_refuse("code_required")
            return

        invite = await roster.consume_invite(code, user_id)
        if not invite:
            self._join_refuse("code_invalid")
            return

        await self._pin_and_admit(
            # The name comes from the invitation, not from the token: the hub does
            # not put a username claim in a JWT, so pinning from the session alone
            # left the roster nameless and `member revoke <name>` unable to match.
            roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64,
            group_id=invite["group_id"], role=invite["role"],
            approved_by=invite["created_by"], via="code")
        # The roster row comes from the invitation; the key comes from the
        # connection. An operator pairing is node-wide (empty group), but they
        # redeemed the code while opening a group and expect to read it — and
        # is_authorized() already grants an operator every group on this node.
        await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"],
                            role=invite["role"], recognised=False)


    # ── Device linking ───────────────────────────────────────────────────────
    #
    # A person may hold several devices on one node. The authority admitting a
    # new one is a key the node already pinned — never the hub, which has stored
    # no user keys since 2026-08-14 and therefore cannot countersign anything.
    # See docs/desktop-client-v1.md §4.

    async def _do_device_request(self, msg: dict) -> None:
        """
        A new device files itself as pending, bound to a code it displays.

        Served in the pre-proof window: by construction the caller holds no key
        this node knows, so there is nothing yet to prove. Filing is inert —
        nothing is admitted until an existing device countersigns.
        """
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id or not self._nonce_node:
            self._send({"type": "error", "detail": "Not ready for a device request"})
            return

        if not self._spend_device_attempt():
            return

        pk_ed_b64 = str(msg.get("pk_ed25519", ""))
        pk_x_b64 = str(msg.get("pk_x25519", ""))
        code_hash = str(msg.get("code_hash", ""))
        if not (pk_ed_b64 and pk_x_b64 and code_hash):
            self._send({"type": "error", "detail": "Missing device keys or code"})
            return

        # The account must already be known here. Anti-spam rather than a
        # security boundary: the filing key is unpinned by construction, so this
        # bounds the table, not the trust.
        existing = await roster.list_devices(self._user_id)
        if not existing:
            self._send({"type": "error",
                        "detail": "This account has no device on this node yet — "
                                  "an invitation code is what admits the first"})
            return
        if len(existing) >= roster.MAX_DEVICES_PER_USER:
            self._send({"type": "error",
                        "detail": f"Already {len(existing)} devices, which is the "
                                  f"limit. Revoke one first."})
            return

        ts = int(msg.get("ts", 0))
        if abs(time.time() - ts) > DEVICE_TTL:
            self._send({"type": "error", "detail": "Device request expired"})
            return

        transcript = device_request_transcript(
            node_pk_b64=self._node_pk_b64(), user_id=self._user_id,
            pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
            code_hash=code_hash, nonce_node=self._nonce_node, ts=ts)
        try:
            pk_ed = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64))
            sig = base64.b64decode(msg.get("sig", ""))
        except Exception:
            self._send({"type": "error", "detail": "Invalid device key encoding"})
            return
        if not self._verify_sig(pk_ed, transcript, sig):
            # Proof of possession, and nothing more: this says the caller holds
            # the keys, never that they belong to this account.
            self._send({"type": "error", "detail": "Device signature invalid"})
            return

        ttl = int(self._ctx.get("device_request_ttl") or 3600)
        expires = await roster.file_device_request(
            user_id=self._user_id, username=self._username or "",
            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64,
            code_hash=code_hash, ttl=ttl)
        self._audit("device_request", f"{pk_ed_b64[:16]}")
        log.info("Device request filed for %s (%s)", self._user_id[:8],
                 pk_ed_b64[:16])
        self._send({"type": MNP.DEVICE_REQUEST_ACK, "v": MNP_VERSION,
                    "expires_at": expires})

    async def _do_device_lookup(self, msg: dict) -> None:
        """
        List this account's pending device requests, each with its code hash.

        **The node never learns the code**, which is what makes it unable to
        substitute a key. It answers with candidates; the approver recomputes
        `sha256(code ‖ keys)` for each and keeps the one that matches. A node
        offering fabricated keys would have to produce a hash matching
        `sha256(code ‖ fabricated)` — and it does not know the code.

        An earlier version of this took the hash from the client and looked the
        request up by it. That is circular: the client cannot compute the hash
        without already knowing the keys it is asking about.
        """
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id:
            self._send({"type": "error", "detail": "Roster not available"})
            return

        pending = await roster.list_device_requests(self._user_id)
        self._send({
            "type": MNP.DEVICE_LOOKUP_RESULT, "v": MNP_VERSION,
            "requests": [
                {"pk_ed25519": r["pk_ed25519"], "pk_x25519": r["pk_x25519"],
                 "code_hash": r["code_hash"], "created_at": r["created_at"]}
                for r in pending
            ],
        })

    async def _do_device_add(self, msg: dict) -> None:
        """
        Admit a device, countersigned by one this node already pinned.

        The whole control is in `_verify_device_signer`: the signature must
        verify against a **live device of this same account**. The hub holds no
        user keys and so cannot produce one.
        """
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id or not self._nonce_node:
            self._send({"type": "error", "detail": "Not ready to add a device"})
            return
        if not self._spend_device_attempt():
            return

        pk_ed_b64 = str(msg.get("pk_ed25519", ""))
        pk_x_b64 = str(msg.get("pk_x25519", ""))
        ts = int(msg.get("ts", 0))
        if not (pk_ed_b64 and pk_x_b64):
            self._send({"type": "error", "detail": "Missing device keys"})
            return
        if abs(time.time() - ts) > DEVICE_TTL:
            self._send({"type": "error", "detail": "Approval expired"})
            return

        transcript = device_add_transcript(
            node_pk_b64=self._node_pk_b64(), user_id=self._user_id,
            pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64,
            nonce_node=self._nonce_node, ts=ts)
        signer = await self._verify_device_signer(roster, transcript,
                                                  msg.get("sig", ""))
        if signer is None:
            self._audit("device_add_refused", pk_ed_b64[:16])
            self._send({"type": "error",
                        "detail": "Not signed by a device already paired here"})
            return

        devices = await roster.list_devices(self._user_id)
        if len(devices) >= roster.MAX_DEVICES_PER_USER:
            self._send({"type": "error", "detail": "Device limit reached"})
            return

        # Spend the request. Single use: an approval cannot be replayed, and a
        # code that was used is gone whatever else happens next.
        code_hash = str(msg.get("code_hash", ""))
        if code_hash and not await roster.take_device_request(
                code_hash, self._user_id):
            self._send({"type": "error",
                        "detail": "That request is no longer pending"})
            return

        await roster.pin_identity(
            user_id=self._user_id, username=self._username or "",
            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via="device",
            label=str(msg.get("label", ""))[:64], added_by_pk=signer)
        self._audit("device_added", f"{pk_ed_b64[:16]} by {signer[:16]}")
        log.info("Device added for %s: %s (approved by %s)",
                 self._user_id[:8], pk_ed_b64[:16], signer[:16])
        self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION,
                    "pk_ed25519": pk_ed_b64})

    async def _do_device_list(self, msg: dict) -> None:
        """This account's devices. Anyone may read their own, nobody else's."""
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id:
            self._send({"type": "error", "detail": "Roster not available"})
            return
        devices = await roster.list_devices(self._user_id)
        pending = await roster.pending_device_requests(self._user_id)
        self._send({
            "type": MNP.DEVICE_LIST_RESULT, "v": MNP_VERSION,
            "pending": pending,
            "devices": [
                {"pk_ed25519": d["pk_ed25519"], "label": d.get("label", ""),
                 "pinned_at": d["pinned_at"], "pinned_via": d["pinned_via"],
                 "added_by_pk": d.get("added_by_pk", ""),
                 "is_this_one": d["pk_ed25519"] == self._pinned_pk}
                for d in devices
            ],
        })

    async def _do_device_revoke(self, msg: dict) -> None:
        """
        Retire one of this account's devices — a lost laptop.

        Countersigned like an addition, by a live device of the same account.
        The last one cannot go: an account with no device on this node can only
        return through an operator's invitation code, and doing that to yourself
        by accident is not a mistake worth allowing.
        """
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id or not self._nonce_node:
            self._send({"type": "error", "detail": "Not ready"})
            return
        if not self._spend_device_attempt():
            return

        target = str(msg.get("pk_ed25519", ""))
        ts = int(msg.get("ts", 0))
        if not target:
            self._send({"type": "error", "detail": "Missing device key"})
            return
        if abs(time.time() - ts) > DEVICE_TTL:
            self._send({"type": "error", "detail": "Request expired"})
            return

        victim = await roster.find_device(self._user_id, target)
        if victim is None:
            self._send({"type": "error", "detail": "No such device"})
            return

        transcript = device_add_transcript(
            node_pk_b64=self._node_pk_b64(), user_id=self._user_id,
            pk_ed25519_b64=target, pk_x25519_b64=victim["pk_x25519"],
            nonce_node=self._nonce_node, ts=ts)
        signer = await self._verify_device_signer(roster, transcript,
                                                  msg.get("sig", ""))
        if signer is None:
            self._send({"type": "error",
                        "detail": "Not signed by a device already paired here"})
            return

        if len(await roster.list_devices(self._user_id)) <= 1:
            self._send({"type": "error",
                        "detail": "This is your only device here — removing it "
                                  "would need an operator code to come back"})
            return

        await roster.revoke_device(self._user_id, target)
        self._audit("device_revoked", f"{target[:16]} by {signer[:16]}")
        log.info("Device revoked for %s: %s", self._user_id[:8], target[:16])
        self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION,
                    "revoked": target})

    async def _verify_device_signer(self, roster, transcript: bytes,
                                    sig_b64: str) -> str | None:
        """
        The pinned key that signed this, or None.

        Every live device of the account is tried, because any of them may
        approve. A revoked one is not in the list — that is the point of marking
        rather than deleting: a lost laptop must stop being able to admit its
        replacement.
        """
        try:
            sig = base64.b64decode(sig_b64)
        except Exception:
            return None
        for device in await roster.list_devices(self._user_id):
            try:
                pk = Ed25519PublicKey.from_public_bytes(
                    base64.b64decode(device["pk_ed25519"]))
            except Exception:
                continue
            if self._verify_sig(pk, transcript, sig):
                return device["pk_ed25519"]
        return None

    def _spend_device_attempt(self) -> bool:
        """
        Bound guessing on this connection, as the join path does.

        A code is 40 bits, single use and bound to the keys it names, so this is
        depth rather than the control — but an unbounded loop over the lookup is
        still a free oracle, and a burst of failures belongs in the audit log.
        """
        self._device_attempts = getattr(self, "_device_attempts", 0) + 1
        if self._device_attempts > 5:
            self._audit("device_attempts_exceeded", str(self._device_attempts))
            self._send({"type": "error",
                        "detail": "Too many device attempts on this connection"})
            return False
        return True

    def _group_join_policy(self, group_id: str) -> str:
        """
        Admission policy for a group, read from the node's own configuration.

        Never from the hub: a hub that could declare a group open would be handed
        the key to it (§3.4 of docs/invite-pairing-v1.md).
        """
        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
        return gctx.get("join_policy", "invite")

    async def _pin_and_admit(
        self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str,
        *, group_id: str, role: str, approved_by: str, via: str,
    ) -> None:
        await roster.pin_identity(
            user_id=user_id, username=username,
            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via,
        )
        await roster.set_member(
            group_id=group_id, user_id=user_id, role=role,
            status="active", approved_by=approved_by,
        )
        if role == ROLE_OPERATOR:
            self._ctx["has_admin_authority"] = True

        log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role)
        self._audit_join("join_pinned", f"role={role} via={via}")

    async def _join_ok(
        self, user_id: str, pk_x_raw: bytes, group_id: str,
        *, role: str, recognised: bool,
    ) -> None:
        """
        Answer a join, wrapping the group key for the key the caller just proved.

        This is the H3 fix. The inviter used to fetch the invitee's public key from
        the hub and wrap the GEK for whatever came back, so a hub that answered
        with its own key was handed the group key by an honest member following the
        protocol exactly. The node now wraps for a key that arrived from its owner
        over an authenticated channel, bound to a pinned identity.
        """
        reply = {
            "type": MNP.JOIN_RESULT,
            "v": MNP_VERSION,
            "ok": True,
            "recognised": recognised,
            "role": role,
        }

        roster = self._ctx["roster"]
        if group_id and not await roster.is_authorized(group_id, user_id):
            # Pinned on this node, but not admitted to this group. Hub membership
            # alone must not produce a key.
            reply["gek"] = False
            reply["reason"] = "not_authorized_for_group"
            self._send(reply)
            self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized")
            return

        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
        gek = gctx.get("gek")
        if not gek:
            reply["gek"] = False
            reply["reason"] = "no_gek"
            self._send(reply)
            return

        bundle = wrap_gek_aes(gek, pk_x_raw)
        reply["gek"] = True
        reply["pk_eph_b64"] = bundle["pk_eph_b64"]
        reply["nonce_b64"] = bundle["nonce_b64"]
        reply["wrapped_b64"] = bundle["wrapped_b64"]
        self._send(reply)
        self._audit_join("gek_wrapped", f"group={group_id[:8]}")

    def _do_dir_create(self, msg: dict) -> None:
        """
        Create a directory, for any member of the group.

        Same confinement as an upload: every segment passes the name allowlist and
        the result must resolve under the shared root. Making a directory is not a
        privileged act — a member who can add a file can organise where it goes —
        but it writes to the operator's disk, so it is audited like one.
        """
        ctx = self._group_ctx()
        roots: RootSet | None = ctx.get("roots")
        if not roots:
            self._send({"type": "error", "detail": "No shared directory"})
            return

        name = str(msg.get("name", "")).strip()
        if not SAFE_UPLOAD_NAME.match(name):
            self._send({"type": "error", "detail": "Invalid directory name"})
            return

        # The virtual root is not a directory on anyone's disk, so a member
        # cannot create one there — that would be adding a root, which is the
        # operator's configuration and not a file operation.
        parent_rel = (msg.get("dir") or "").strip("/")
        if not parent_rel:
            self._send({"type": "error",
                        "detail": "Choose a folder to create this in"})
            return

        parent = safe_subdir(roots, parent_rel)
        if parent is None or not parent.is_dir():
            self._send({"type": "error", "detail": "Invalid directory"})
            return

        target = safe_subdir(roots, f"{parent_rel}/{name}")
        if target is None:
            self._send({"type": "error", "detail": "Invalid directory"})
            return
        if target.exists():
            self._send({"type": "error", "detail": "Already exists"})
            return

        target.mkdir(parents=False)
        virtual = roots.virtual_of(target) or f"{parent_rel}/{name}"
        log.info("Directory created by %s: %s", self._user_id[:8], virtual)
        self._audit("dir_create", virtual)
        self._send({
            "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION,
            "dir": virtual,
        })

    @staticmethod
    def _names_a_root(roots: RootSet, rel: str) -> bool:
        """True when `rel` is a bare root name rather than something inside one."""
        found = roots.split(rel or "")
        return found is not None and not found[1]

    def _do_dir_delete(self, msg: dict) -> None:
        """
        Remove an empty directory, for the node operator.

        Creating one is not privileged — a member who can add a file may organise
        where it goes — but removing one is: it acts on a name other members are
        using, and on the operator's disk. Empty is the whole safety property
        here. Nothing recursive: refusing a directory with anything in it means
        this can never destroy content, whatever the caller intended, so the
        operator deletes the files first and sees what they are losing.
        """
        ctx = self._group_ctx()
        roots: RootSet | None = ctx.get("roots")
        if not roots:
            self._send({"type": "error", "detail": "No shared directory"})
            return

        rel = (msg.get("dir") or "").strip("/")
        target = safe_subdir(roots, rel)
        # A root itself is not deletable here: removing one is a configuration
        # change, and doing it through a file operation would leave the group
        # config naming a directory nobody can reach.
        if target is None or self._names_a_root(roots, rel):
            self._send({"type": "error", "detail": "Invalid directory"})
            return
        if not target.is_dir():
            self._send({"type": "error", "detail": "Not a directory"})
            return
        if any(target.iterdir()):
            self._send({"type": "error", "detail": "Directory is not empty"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for deletion"})
            return

        self._issue_admin_challenge(
            OP_DIR_DELETE, roots.virtual_of(target) or rel)

    async def _admin_exec_dir_delete(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        rel = pending["subject"]
        ctx = self._group_ctx()
        roots: RootSet | None = ctx.get("roots")
        target = safe_subdir(roots, rel) if roots else None
        if (target is None or self._names_a_root(roots, rel)
                or not target.is_dir()):
            self._send({"type": "error", "detail": "Not a directory"})
            return

        # Operator only. A file has an uploader who may remove their own; a
        # directory has none, so there is no second key to accept here.
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"dir_delete:{rel}")
            return

        # Checked again after the signature: the emptiness test that let this
        # through happened before a round trip to the operator's browser, and a
        # file could have landed in the meantime.
        if any(target.iterdir()):
            self._send({"type": "error", "detail": "Directory is not empty"})
            return

        target.rmdir()
        log.info("Directory removed by %s: %s", self._user_id[:8], rel)
        self._audit("dir_delete", rel)
        self._send({"type": MNP.DIR_DELETE_ACK, "v": MNP_VERSION, "dir": rel})

    def _do_member_revoke(self, msg: dict) -> None:
        """
        Stop serving the group key to someone, at the operator's request.

        The same authority as an invite, and the same reason: the roster decides
        who this node serves, so only a key the node pinned as an operator may
        change it. Membership on the hub is not consulted — the hub can remove
        someone from a group, and that stops them reaching the node at all, but
        it cannot make the node forget them.
        """
        user_id = str(msg.get("user_id", "")).strip()
        if not user_id:
            self._send({"type": "error", "detail": "Missing user_id"})
            return
        if user_id == self._user_id:
            # Removing yourself from your own node is not a member operation;
            # it would leave the group with nobody able to invite.
            self._send({"type": "error", "detail": "Cannot revoke yourself"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id)

    def _do_gek_rotate(self, msg: dict) -> None:
        """
        Ask for a new group key. Operator only, and signed.

        This is what actually removes a revoked member's access: revocation
        stops the node serving the *next* key, and they still hold the current
        one. The node generates the replacement itself — nothing arriving here
        contributes key material, which is what the C5b rule is about.
        """
        group_id = str(msg.get("group_id", "")).strip() or self._group_id
        if not group_id:
            self._send({"type": "error", "detail": "No group on this connection"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_GEK_ROTATE, group_id, group_id=group_id)

    async def _admin_exec_gek_rotate(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"gek_rotate:{pending['subject'][:8]}")
            return
        try:
            result = await self._run_op(
                ops.set_gek, pending["subject"], rotate=True)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("gek_rotate", pending["subject"])
        self._send({
            "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION,
            "group_id": pending["subject"],
            "authorized_members": result.get("authorized_members", 0),
            # Said plainly, because rotating is the step people skip: content
            # already downloaded stays readable to whoever holds it.
            "note": "members re-receive the key on their next connect; content "
                    "already downloaded is unaffected",
        })

    def _do_member_unpin(self, msg: dict) -> None:
        """Forget a pinned identity, so someone can pair again with a new key."""
        user_id = str(msg.get("user_id", "")).strip()
        if not user_id:
            self._send({"type": "error", "detail": "Missing user_id"})
            return
        if user_id == self._user_id:
            # Unpinning yourself over the connection your pin authorizes would
            # end that connection's authority mid-operation.
            self._send({"type": "error", "detail": "Cannot unpin yourself"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id)

    async def _admin_exec_member_unpin(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        user_id = pending["subject"]
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"member_unpin:{user_id[:8]}")
            return
        try:
            await self._run_op(ops.unpin_member, user_id)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("member_unpin", user_id)
        self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION,
                    "user_id": user_id})

    def _do_member_upload(self, msg: dict) -> None:
        # Deprecated: upload control is now per-root via writable flag.
        # Old clients may still send this — acknowledge without acting.
        log.warning("Deprecated member_upload message received — use root "
                    "writable/read-only instead")
        self._send({"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
                    "allowed": True, "deprecated": True})

    # Every "application" a group can show. Photos joins this set (and
    # apps.js's registry, client-side) when it lands; nothing else about
    # this handler changes. DEFAULT_APPS (roster.py) deliberately does not
    # include "video" or "music" — both can make outbound third-party
    # network calls (TMDB, MusicBrainz) once enabled, so an operator opts a
    # group in explicitly rather than getting it for free
    # (docs/mediacenter.md §5.6, docs/musicbay.md §4.4).
    ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo"})

    def _do_apps_enabled(self, msg: dict) -> None:
        """
        Turn a group "application" on or off for everyone, for this group.

        Signed like the root ops: this decides what a member sees, and an
        unsigned message would let any member turn a disabled one back on.
        """
        apps = msg.get("apps")
        if not isinstance(apps, list) or not apps:
            self._send({"type": "error", "detail": "Missing or empty apps"})
            return
        unknown = set(apps) - self.ALLOWED_APPS
        if unknown:
            self._send({"type": "error",
                       "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"})
            return
        # Files is not a toggle: MNP permits root exploration regardless of
        # what this list says, so hiding the tab only ever misled. Added at the
        # front, the same order ops.set_enabled_apps writes, so the landing-tab
        # preference sees one list and not two.
        if "files" not in apps:
            apps.insert(0, "files")
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        # The subject is what the operator is shown before signing, and what
        # the client compares its own request against (transport.js) — a
        # canonical form so both sides build the same transcript.
        self._issue_admin_challenge(OP_APPS_ENABLED, ",".join(sorted(apps)))

    async def _admin_exec_apps_enabled(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        apps = pending["subject"].split(",") if pending["subject"] else []
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"apps_enabled:{pending['subject']}")
            return
        try:
            await self._run_op(
                ops.set_enabled_apps, self._group_id or "", apps)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("apps_enabled", pending["subject"])

        # Everyone already connected is told, so a disabled tab disappears
        # without waiting for a reconnection.
        notice = {"type": MNP.APPS_ENABLED_ACK, "v": MNP_VERSION, "apps": apps}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_tmdb_config(self, msg: dict) -> None:
        """
        Optionally set (or clear) a custom TMDB API token, and optionally set
        the language TMDB is queried in (e.g. "fr-FR") — one for the whole
        node, since both are one operator's shared credential/cache, not a
        per-group concern (see _do_tmdb_enabled for the per-group on/off
        switch). Signed like the rest: this changes outbound third-party
        network traffic the node did not have before the Videos app
        (docs/mediacenter.md §5.5, §8) — an unsigned change would let any
        member alter egress the operator never agreed to.
        """
        token = msg.get("token")
        if token is not None and not isinstance(token, str):
            self._send({"type": "error", "detail": "Invalid 'token'"})
            return
        language = msg.get("language")
        if language is not None and not isinstance(language, str):
            self._send({"type": "error", "detail": "Invalid 'language'"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        # The subject is the signed, audited, human-shown string — it must
        # never contain the token itself (it would end up in the audit log
        # in plaintext). The actual token travels only in `payload`, which
        # is node-side context, never re-sent or re-verified from the wire.
        # The language is not a secret, so it travels in the subject itself.
        subject = f"custom_token={'yes' if token else 'no'},language={language or 'default'}"
        self._issue_admin_challenge(
            OP_TMDB_CONFIG, subject,
            payload={"token": token, "language": language},
            group_id="")

    async def _admin_exec_tmdb_config(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"tmdb_config:{pending['subject']}")
            return
        p = pending.get("payload") or {}
        try:
            result = await self._run_op(
                ops.set_tmdb_config, p.get("token"), p.get("language"))
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("tmdb_config", pending["subject"])

        # Node-wide setting: every connected peer in every group is told, not
        # just this group's peers (unlike apps_enabled/the root ops/the
        # per-group tmdb_enabled below).
        notice = {
            "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION,
            "token_customized": result["token_customized"],
            "language": result["language"],
        }
        for gctx in self._ctx.get("groups", {}).values():
            for session in list(gctx.get("_peers", {}).values()):
                try:
                    session._send(notice)
                except Exception:
                    pass

    def _do_tmdb_enabled(self, msg: dict) -> None:
        """
        Whether TMDB lookups run for this group at all. Per-group, unlike
        tmdb_config's token/language — see ops.set_tmdb_enabled. Signed like
        video_root: it decides whether this group's members' Videos tab ever
        makes outbound TMDB traffic.
        """
        enabled = msg.get("enabled")
        if not isinstance(enabled, bool):
            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_TMDB_ENABLED, str(enabled))

    async def _admin_exec_tmdb_enabled(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        enabled = pending["subject"] == "True"
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"tmdb_enabled:{pending['subject']}")
            return
        try:
            await self._run_op(ops.set_tmdb_enabled, self._group_id or "", enabled)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("tmdb_enabled", pending["subject"])

        notice = {"type": MNP.TMDB_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_video_root(self, msg: dict) -> None:
        """
        Which folder (possibly a subfolder of a shared root) the Videos app
        treats as its entry point for this group. Signed like apps_enabled:
        it decides what every member's Videos tab shows.

        An empty path is always accepted (it means "the whole group index",
        today's behaviour). A non-empty path must resolve to a real,
        currently-readable directory — validated against the group's own
        roots the same way directory creation/deletion already is, so a
        stale or mistyped path is refused before a signature is even asked
        for.
        """
        path = msg.get("path")
        if not isinstance(path, str):
            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
            return
        path = path.strip("/")
        if path:
            ctx = self._group_ctx()
            resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None
            if not resolved or not resolved.is_dir():
                self._send({"type": "error", "detail": "Not a directory in this group"})
                return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_VIDEO_ROOT, path)

    async def _admin_exec_video_root(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        path = pending["subject"]
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"video_root:{path}")
            return
        try:
            await self._run_op(ops.set_video_root, self._group_id or "", path)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("video_root", path)

        notice = {"type": MNP.VIDEO_ROOT_ACK, "v": MNP_VERSION, "path": path}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_audio_root(self, msg: dict) -> None:
        """Same shape as _do_video_root above — the Music app's own entry point."""
        path = msg.get("path")
        log.debug("audio_root request user=%s path=%r",
                 (self._user_id or "?")[:8], path)
        if not isinstance(path, str):
            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
            return
        path = path.strip("/")
        if path:
            ctx = self._group_ctx()
            resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None
            if not resolved or not resolved.is_dir():
                self._send({"type": "error", "detail": "Not a directory in this group"})
                return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_AUDIO_ROOT, path)

    async def _admin_exec_audio_root(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        path = pending["subject"]
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"audio_root:{path}")
            return
        try:
            await self._run_op(ops.set_audio_root, self._group_id or "", path)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("audio_root", path)

        notice = {"type": MNP.AUDIO_ROOT_ACK, "v": MNP_VERSION, "path": path}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_photo_roots(self, msg: dict) -> None:
        """
        Which folder(s) the Photos app treats as its entry points for this
        group (docs/photos.md §2.1) — a *set*, replaced whole in one signed
        op, same shape as apps_enabled rather than one op per root the way
        video_root/audio_root are single values.

        An empty list is always accepted (nothing configured yet, today's
        "Photos shows nothing" state). Every non-empty path must resolve to
        a real, currently-readable directory, and no root may be nested
        inside another in the same submitted set — both checked, and
        refused, before a signature is ever asked for, same principle as
        video_root's path check and apps_enabled's "empty set refused up
        front".
        """
        roots = msg.get("roots")
        if not isinstance(roots, list) or not all(isinstance(r, str) for r in roots):
            self._send({"type": "error", "detail": "Missing or invalid 'roots'"})
            return
        roots = sorted({r.strip("/") for r in roots if r.strip("/")})
        ctx = self._group_ctx()
        for path in roots:
            resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None
            if not resolved or not resolved.is_dir():
                self._send({"type": "error",
                           "detail": f"Not a directory in this group: {path}"})
                return
        # Case-insensitive nesting check (§6.8) — a root may not be a folder
        # itself sitting inside another root in the same set.
        folded = [r.casefold() for r in roots]
        for i, a in enumerate(folded):
            for j, b in enumerate(folded):
                if i != j and (a == b or a.startswith(b + "/")):
                    self._send({"type": "error",
                               "detail": f"Root nested inside another: {roots[i]}"})
                    return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_PHOTO_ROOTS, ",".join(roots))

    async def _admin_exec_photo_roots(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        roots = pending["subject"].split(",") if pending["subject"] else []
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"photo_roots:{pending['subject']}")
            return
        try:
            await self._run_op(ops.set_photo_roots, self._group_id or "", roots)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("photo_roots", pending["subject"])

        notice = {"type": MNP.PHOTO_ROOTS_ACK, "v": MNP_VERSION, "roots": roots}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_musicbrainz_enabled(self, msg: dict) -> None:
        """
        Whether MusicBrainz lookups run for this group at all. Per-group
        from the start (docs/musicbay.md §3.2/§6) — signed like
        tmdb_enabled/video_root: it decides whether this group's members'
        Music tab ever makes outbound MusicBrainz traffic.
        """
        enabled = msg.get("enabled")
        if not isinstance(enabled, bool):
            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_MUSICBRAINZ_ENABLED, str(enabled))

    async def _admin_exec_musicbrainz_enabled(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        enabled = pending["subject"] == "True"
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"musicbrainz_enabled:{pending['subject']}")
            return
        try:
            await self._run_op(ops.set_musicbrainz_enabled, self._group_id or "", enabled)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("musicbrainz_enabled", pending["subject"])

        notice = {"type": MNP.MUSICBRAINZ_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    # Reconcile's backstop and the watchdog debounce (indexer.py
    # DirectoryIndexer) — how hard the node works on the operator's own
    # disk, not a member-facing permission. Signed for the same reason as
    # apps_enabled: consistency of the authorization model, not because a
    # wrong value here is itself dangerous.
    MIN_RECONCILE_SECS = 10.0
    MAX_RECONCILE_SECS = 24 * 3600.0
    MIN_DEBOUNCE_SECS = 0.0
    MAX_DEBOUNCE_SECS = 300.0

    def _do_set_scan_settings(self, msg: dict) -> None:
        try:
            reconcile = float(msg.get("reconcile_interval_secs"))
            debounce = float(msg.get("debounce_secs"))
        except (TypeError, ValueError):
            self._send({"type": "error", "detail": "Invalid scan settings"})
            return
        if not (self.MIN_RECONCILE_SECS <= reconcile <= self.MAX_RECONCILE_SECS):
            self._send({"type": "error",
                       "detail": f"reconcile_interval_secs must be between "
                                 f"{self.MIN_RECONCILE_SECS:.0f} and "
                                 f"{self.MAX_RECONCILE_SECS:.0f}"})
            return
        if not (self.MIN_DEBOUNCE_SECS <= debounce <= self.MAX_DEBOUNCE_SECS):
            self._send({"type": "error",
                       "detail": f"debounce_secs must be between "
                                 f"{self.MIN_DEBOUNCE_SECS:.0f} and "
                                 f"{self.MAX_DEBOUNCE_SECS:.0f}"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(
            OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}")

    async def _admin_exec_set_scan_settings(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        try:
            reconcile_s, debounce_s = pending["subject"].split(",")
            reconcile, debounce = float(reconcile_s), float(debounce_s)
        except (ValueError, KeyError):
            self._send({"type": "error", "detail": "Invalid scan settings"})
            return
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"set_scan_settings:{pending['subject']}")
            return
        try:
            result = await self._run_op(
                ops.set_scan_settings, self._group_id or "", reconcile, debounce)
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("set_scan_settings", pending["subject"])

        notice = {"type": MNP.SET_SCAN_SETTINGS_ACK, "v": MNP_VERSION, **result}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    # ── Node management (D5) ─────────────────────────────────────────────────

    async def _do_node_status(self, msg: dict) -> None:
        """All groups, roots, peers — the operator's overview."""
        node_uid = self._ctx.get("node_user_id")
        log.info("node_status: user=%s node_user=%s admin=%s",
                 self._user_id, node_uid, self._is_node_admin())
        if not self._is_node_admin():
            self._send({"type": "error", "detail": "Not the node operator"})
            return
        try:
            result = await self._run_op(ops.list_groups)
            self._send({"type": MNP.NODE_STATUS_ACK, "v": MNP_VERSION, **result})
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
        except Exception as e:
            log.error("node_status failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})

    async def _do_node_settings_set(self, msg: dict) -> None:
        if not self._is_node_admin():
            self._send({"type": "error", "detail": "Not the node operator"})
            return
        settings = msg.get("settings", {})
        if not settings:
            self._send({"type": "error", "detail": "No settings provided"})
            return
        try:
            result = await self._run_op(ops.set_node_settings, settings)
            self._send({"type": MNP.NODE_SETTINGS_SET_ACK, "v": MNP_VERSION,
                         **result})
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
        except Exception as e:
            log.error("node_settings_set failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})

    async def _do_roster_read(self, msg: dict) -> None:
        if not self._is_node_admin():
            self._send({"type": "error", "detail": "Not the node operator"})
            return
        group_id = str(msg.get("group_id", "")).strip()
        try:
            result = await self._run_op(ops.read_roster, group_id)
            self._send({"type": MNP.ROSTER_READ_ACK, "v": MNP_VERSION, **result})
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
        except Exception as e:
            log.error("roster_read failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})

    async def _do_denylist_read(self, msg: dict) -> None:
        if not self._is_node_admin():
            self._send({"type": "error", "detail": "Not the node operator"})
            return
        try:
            result = await self._run_op(ops.read_denylist)
            self._send({"type": MNP.DENYLIST_READ_ACK, "v": MNP_VERSION, **result})
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
        except Exception as e:
            log.error("denylist_read failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})

    async def _do_denylist_clear(self, msg: dict) -> None:
        if not self._is_node_admin():
            self._send({"type": "error", "detail": "Not the node operator"})
            return
        subject = str(msg.get("subject", "")).strip()
        try:
            result = await self._run_op(ops.clear_denylist, subject=subject)
            self._audit("denylist_clear", subject or "all")
            self._send({"type": MNP.DENYLIST_CLEAR_ACK, "v": MNP_VERSION, **result})
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
        except Exception as e:
            log.error("denylist_clear failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})

    def _do_group_attach(self, msg: dict) -> None:
        name = str(msg.get("name", "")).strip()
        shared_dir = str(msg.get("shared_dir", "")).strip()
        if not name or not shared_dir:
            self._send({"type": "error", "detail": "Missing name or shared_dir"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        upload_dir = str(msg.get("upload_dir", "")).strip()
        self._issue_admin_challenge(
            OP_GROUP_ATTACH, name,
            payload={"name": name, "shared_dir": shared_dir,
                     "upload_dir": upload_dir},
            group_id="")

    async def _admin_exec_group_attach(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed",
                        f"group_attach:{pending['subject'][:16]}")
            return
        p = pending.get("payload") or {}
        try:
            result = await self._run_op(
                ops.attach_group, p["name"], p["shared_dir"], p.get("upload_dir", ""))
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("group_attach", pending["subject"])
        self._send({"type": MNP.GROUP_ATTACH_ACK, "v": MNP_VERSION, **result})
        state = self._ctx.get("daemon_state")
        reload_fn = state.get("reload_fn") if state else None
        if reload_fn:
            try:
                await reload_fn()
            except Exception as e:
                log.error("Reload after group_attach failed: %s", e)

    def _do_group_detach(self, msg: dict) -> None:
        name = str(msg.get("name", "")).strip()
        if not name:
            self._send({"type": "error", "detail": "Missing group name or id"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(
            OP_GROUP_DETACH, name,
            payload={"name": name},
            group_id="")

    async def _admin_exec_group_detach(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed",
                        f"group_detach:{pending['subject'][:16]}")
            return
        p = pending.get("payload") or {}
        try:
            result = await self._run_op(ops.detach_group, p["name"])
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        self._audit("group_detach", pending["subject"])
        self._send({"type": MNP.GROUP_DETACH_ACK, "v": MNP_VERSION, **result})
        state = self._ctx.get("daemon_state")
        reload_fn = state.get("reload_fn") if state else None
        if reload_fn:
            try:
                await reload_fn()
            except Exception as e:
                log.error("Reload after group_detach failed: %s", e)

    async def _do_node_reload(self, msg: dict) -> None:
        if not self._is_node_admin():
            self._send({"type": "error", "detail": "Not the node operator"})
            return
        state = self._ctx.get("daemon_state")
        reload_fn = state.get("reload_fn") if state else None
        if not reload_fn:
            self._send({"type": "error", "detail": "Reload not available"})
            return
        try:
            await reload_fn()
            self._send({"type": MNP.NODE_RELOAD_ACK, "v": MNP_VERSION,
                        "status": "reloaded"})
        except Exception as e:
            log.error("node_reload failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Reload failed"})

    def _do_root_add(self, msg: dict) -> None:
        target_group = str(msg.get("group_id", "")).strip()
        path = str(msg.get("path", "")).strip()
        if not target_group or not path:
            self._send({"type": "error", "detail": "Missing group_id or path"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(
            OP_ROOT_ADD, path,
            payload={
                "group_id": target_group, "path": path,
                "name": str(msg.get("name", ""))[:128],
                "kind": str(msg.get("kind", "generic"))[:16],
                "writable": bool(msg.get("writable", msg.get("upload", False))),
                "removable": bool(msg.get("removable", False)),
            },
            group_id=target_group)

    async def _admin_exec_root_add(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"root_add:{pending['subject'][:24]}")
            return
        p = pending["payload"]
        try:
            result = await self._run_op(
                ops.add_root, p["group_id"], p["path"],
                name=p.get("name", ""), kind=p.get("kind", "generic"),
                writable=p.get("writable", False),
                removable=p.get("removable", False))
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        except Exception as e:
            log.error("root_add failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})
            return
        self._audit("root_add", f"{p['path']}→{p['group_id'][:8]}")
        await self._retarget_indexer(p["group_id"])
        self._send({"type": MNP.ROOT_ADD_ACK, "v": MNP_VERSION, **result})

    def _do_root_remove(self, msg: dict) -> None:
        target_group = str(msg.get("group_id", "")).strip()
        root_name = str(msg.get("root_name", "")).strip()
        if not target_group or not root_name:
            self._send({"type": "error", "detail": "Missing group_id or root_name"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(
            OP_ROOT_REMOVE, root_name,
            payload={"group_id": target_group, "root_name": root_name},
            group_id=target_group)

    async def _admin_exec_root_remove(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"root_remove:{pending['subject'][:24]}")
            return
        p = pending["payload"]
        try:
            result = await self._run_op(
                ops.remove_root, p["group_id"], p["root_name"])
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        except Exception as e:
            log.error("root_remove failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})
            return
        self._audit("root_remove", f"{p['root_name']}←{p['group_id'][:8]}")
        await self._retarget_indexer(p["group_id"])
        self._send({"type": MNP.ROOT_REMOVE_ACK, "v": MNP_VERSION, **result})

    def _do_root_update(self, msg: dict) -> None:
        target_group = str(msg.get("group_id", self._group_id or "")).strip()
        root_name = str(msg.get("root_name", "")).strip()
        if not target_group or not root_name:
            self._send({"type": "error", "detail": "Missing group_id or root_name"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        updates = []
        if "writable" in msg:
            updates.append(f"rw={'on' if msg['writable'] else 'off'}")
        if "removable" in msg:
            updates.append(f"rem={'on' if msg['removable'] else 'off'}")
        subject = f"{root_name}:{','.join(updates)}" if updates else root_name
        self._issue_admin_challenge(
            OP_ROOT_UPDATE, subject,
            payload={
                "group_id": target_group, "root_name": root_name,
                "writable": msg.get("writable"),
                "removable": msg.get("removable"),
            },
            group_id=target_group)

    async def _admin_exec_root_update(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed",
                        f"root_update:{pending['subject'][:24]}")
            return
        p = pending["payload"]
        try:
            result = await self._run_op(
                ops.update_root, p["group_id"], p["root_name"],
                writable=p.get("writable"), removable=p.get("removable"))
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        except Exception as e:
            log.error("root_update failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})
            return
        self._audit("root_update", pending["subject"])
        await self._retarget_indexer(p["group_id"])
        notice = {"type": MNP.ROOT_UPDATE_ACK, "v": MNP_VERSION, **result}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_root_eject(self, msg: dict) -> None:
        target_group = str(msg.get("group_id", self._group_id or "")).strip()
        root_name = str(msg.get("root_name", "")).strip()
        if not target_group or not root_name:
            self._send({"type": "error", "detail": "Missing group_id or root_name"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(
            OP_ROOT_EJECT, root_name,
            payload={"group_id": target_group, "root_name": root_name},
            group_id=target_group)

    async def _admin_exec_root_eject(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed",
                        f"root_eject:{pending['subject'][:24]}")
            return
        p = pending["payload"]
        try:
            result = await self._run_op(
                ops.eject_root, p["group_id"], p["root_name"])
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        except Exception as e:
            log.error("root_eject failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})
            return
        self._audit("root_eject", p["root_name"])
        notice = {"type": MNP.ROOT_EJECT_ACK, "v": MNP_VERSION, **result}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_root_plug(self, msg: dict) -> None:
        target_group = str(msg.get("group_id", self._group_id or "")).strip()
        root_name = str(msg.get("root_name", "")).strip()
        if not target_group or not root_name:
            self._send({"type": "error", "detail": "Missing group_id or root_name"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(
            OP_ROOT_PLUG, root_name,
            payload={"group_id": target_group, "root_name": root_name},
            group_id=target_group)

    async def _admin_exec_root_plug(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed",
                        f"root_plug:{pending['subject'][:24]}")
            return
        p = pending["payload"]
        try:
            result = await self._run_op(
                ops.plug_root, p["group_id"], p["root_name"])
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return
        except Exception as e:
            log.error("root_plug failed: %s", e, exc_info=True)
            self._send({"type": "error", "detail": "Internal error"})
            return
        self._audit("root_plug", p["root_name"])
        notice = {"type": MNP.ROOT_PLUG_ACK, "v": MNP_VERSION, **result}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    async def _run_op(self, fn, *args, **kwargs):
        """
        Call an operation from `meshbay_node.ops` with the daemon's own view.

        The transport carries its own context and the loopback API carries the
        daemon state; they overlap but are not the same dict. Handing the MNP
        path a *second* set of lookups is exactly how two implementations of one
        operation start disagreeing — C1 and C6 one size down — so the daemon
        publishes its state here and both adapters call the same function.
        """
        state = self._ctx.get("daemon_state")
        if state is None:
            raise ops.OpError("Node state not available", status=503)
        return await fn(state, *args, **kwargs)

    async def _retarget_indexer(self, group_id: str) -> None:
        """Tell the indexer to rescan after roots changed."""
        state = self._ctx.get("daemon_state")
        if not state:
            return
        indexer = state.get("indexers", {}).get(group_id)
        roots = state.get("groups_ctx", {}).get(group_id, {}).get("roots")
        if indexer and roots:
            await indexer.retarget(roots)

    async def _admin_exec_member_revoke(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        user_id = pending["subject"]
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"member_revoke:{user_id[:8]}")
            return

        try:
            result = await self._run_op(
                ops.revoke_member, user_id, self._group_id or "")
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return

        # Anyone connected right now keeps the key they already unwrapped; what
        # they lose is the next one. Rotating it is the operator's call, and the
        # ack says so rather than implying this undid anything already read.
        peer = self._peer_registry().get(user_id)
        if peer is not None:
            try:
                await peer.close()
            except Exception:
                pass

        self._audit("member_revoke", user_id)
        self._send({
            "type": MNP.MEMBER_REVOKE_ACK, "v": MNP_VERSION,
            "user_id": user_id,
            "reminder": result.get("reminder", ""),
        })

    async def _do_keypair_bundle_delete(self) -> None:
        """
        Withdraw our own key backup from this node.

        Only ever our own: the user_id comes from the authenticated session, never
        from the message. Someone who does not want a second browser should not be
        leaving a PBKDF2-protected blob on every node they have ever joined (C4),
        and turning the setting off has to remove what is already there — not just
        stop adding to it.
        """
        bundle_store = self._ctx.get("bundle_store")
        if not bundle_store:
            self._send({"type": "error", "detail": "Bundle store not available"})
            return

        removed = await bundle_store.delete_keypair(self._user_id)
        if removed:
            log.info("Keypair bundle withdrawn by user=%s", self._user_id[:8])
            self._audit("keypair_bundle_delete")
        self._send({"type": "ack", "v": MNP_VERSION,
                    "detail": "keypair_bundle_deleted", "removed": removed})

    def _audit_pre_proof_fetch(self, mtype: str) -> None:
        """Record bundle access made before the GEK proof (C4)."""
        audit = self._ctx.get("audit_store")
        if not audit:
            return
        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
        self._spawn(audit.log_event(
            user_id=getattr(self, "_pending_sub", "unknown"),
            event="pre_proof_fetch",
            ip=self._remote_ip,
            username=self._username or getattr(self, "_pending_username", ""),
            group_id=getattr(self, "_pending_group", "") or "",
            detail=mtype,
        ))

    def _audit_auth_failed(self, group_id: str, reason: str) -> None:
        audit = self._ctx.get("audit_store")
        if audit:
            self._remote_ip = _get_remote_ip(self._pc)
            self._spawn(audit.log_event(
                user_id="unknown",
                event="auth_failed",
                ip=self._remote_ip,
                group_id=group_id,
                detail=reason,
            ))

    def _spawn(self, coro) -> asyncio.Task:
        """Run a coroutine in the background and hold on to it.

        The reference is what keeps the task alive; the done callback is what
        stops the set growing. Anything that owns a resource for its lifetime —
        a transcode slot, an ffmpeg process — must go through here rather than
        `asyncio.ensure_future`.
        """
        task = asyncio.ensure_future(coro)
        self._tasks.add(task)
        def _on_done(t):
            self._tasks.discard(t)
            if not t.cancelled() and t.exception():
                log.error("Spawned task failed: %s", t.exception(), exc_info=t.exception())
        task.add_done_callback(_on_done)
        return task

    def _group_ctx(self) -> dict:
        if "groups" in self._ctx and self._group_id:
            return self._ctx["groups"][self._group_id]
        return self._ctx

    def _indexing_status(self) -> dict:
        """
        {"scanning": bool, "scanned_bytes": int, "total_bytes": int} for the
        handshake ack and INDEX_PROGRESS pushes — never a path or filename,
        that stays local to the operator's own admin UI. Absent "progress"
        (context not loaded, or a group with no indexer at all) reads as
        idle rather than erroring.
        """
        progress = self._group_ctx().get("progress")
        if progress is None:
            return {"scanning": False, "scanned_bytes": 0, "total_bytes": 0}
        return {
            "scanning": progress.scanning,
            "scanned_bytes": progress.scanned_bytes,
            "total_bytes": progress.total_bytes,
        }

    def _peer_registry(self) -> dict:
        """
        Connected peers for THIS group only.

        Finding H1: this used to live on the shared transport context, so a chat
        message was broadcast to every peer on the node regardless of which group
        they had authenticated to.
        """
        return self._group_ctx().setdefault("_peers", {})

    def _user_names(self) -> dict:
        """Display-name cache, per group — same leak as _peer_registry (H1)."""
        return self._group_ctx().setdefault("_user_names", {})

    def _do_index_sync(self) -> None:
        ctx = self._group_ctx()
        self._send(index_sync_message(ctx["index"], ctx.get("roots")))

    async def _try_serve_thumbnail(
        self, thumb_hash: str, chunk_index: int, gek: bytes | None,
    ) -> dict | None:
        """
        docs/mediacenter.md §5.3: a thumbnail is served through the same
        chunked file_req path as a real file, resolved against the media
        cache instead of the index when the id doesn't match a file.
        Sliced by `chunk_index` like a real file's chunks, not just handed
        back whole: a thumbnail/poster/cover never approached CHUNK_SIZE so
        this used to be equivalent to "only chunk 0 exists", but an audio
        transcode result (docs/musicbay.md, the WMA/Musepack exception) is
        cached in the same media_cache blob store and can be several MB —
        genuinely multi-chunk, same as a file read straight off disk.
        """
        media_cache = self._ctx.get("media_cache")
        if media_cache is None:
            return None
        blob = await media_cache.get_thumb(thumb_hash)
        if blob is None:
            return None
        start = chunk_index * CHUNK_SIZE
        if start > len(blob) or (start == len(blob) and chunk_index != 0):
            return None
        piece = blob[start:start + CHUNK_SIZE]
        return file_chunk_wire(
            gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash)

    async def _do_file_request(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg["file_id"]
        chunk_index = msg["chunk_index"]
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek"))
            if thumb is not None:
                log.debug("file_req file_id=%s chunk=%s: served as thumbnail", file_id[:16], chunk_index)
                self._send(thumb)
                return
            log.warning("File not found: %s", file_id[:16])
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path = entry_abs_path(ctx["roots"], entry)
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        log.debug("dl: req file=%s chunk=%s buffered=%s",
                 file_id[:12], chunk_index,
                 getattr(self._channel, "bufferedAmount", "?"))
        file_hash = bytes.fromhex(entry.id)
        chunk_data = _read_and_encrypt(
            ctx["gek"], file_path, chunk_index, file_hash, entry.id)
        # Backpressure. Without it the node hands the whole window to the
        # channel at once and the reader sees the first chunk, then nothing for
        # as long as the link takes to drain the rest.
        waited = 0.0
        while (self._channel is not None
               and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH
               and self._channel.readyState == "open"
               and waited < 60):
            await asyncio.sleep(0.05)
            waited += 0.05
        if self._channel is None or self._channel.readyState != "open":
            return
        self._send(chunk_data)
        log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s",
                 file_id[:12], chunk_index, len(chunk_data.get("ct") or b""),
                 getattr(self._channel, "bufferedAmount", "?"))
        if chunk_index == 0:
            self._audit("file_download", entry.name)

    @staticmethod
    async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None:
        """
        Downloads a TMDB poster/backdrop once, caches it under its own
        blake3 like a video thumbnail (docs/mediacenter.md §5.4), and
        returns the hash a client then fetches via the normal file_req/
        chunk path (§5.3) — no client ever contacts image.tmdb.org directly.

        Checked by the synthetic `tmdb:{poster_path}` id *before* touching
        the network: without this, every `media_meta_req` for an
        already-cached file re-downloaded the same poster from TMDB (found
        live — a poster grid re-fetched both a show's poster and backdrop
        from TMDB on every single visit, real added latency and needless
        outbound traffic for an image that never changes).
        """
        if not poster_path:
            return None
        synthetic_id = f"tmdb:{poster_path}"
        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
        if cached_hash is not None:
            return cached_hash
        content = await tmdb_client.fetch_image(tmdb_client.poster_url(poster_path))
        if content is None:
            return None
        thumb_hash = blake3.blake3(content).hexdigest()
        await media_cache.put_thumb(thumb_hash, synthetic_id, content)
        return thumb_hash

    @staticmethod
    async def _fetch_and_cache_cover(media_cache, musicbrainz_client, mbid: str | None) -> str | None:
        """
        Music app equivalent of `_fetch_and_cache_poster` — a release's
        Cover Art Archive image, fetched once per mbid and cached under its
        own blake3, addressed the same synthetic-id trick
        (`musicbrainz:{mbid}`) so a second track of the same album never
        re-downloads it. Most releases have no scan at all; that's a normal
        outcome (None), not an error.
        """
        if not mbid:
            return None
        synthetic_id = f"musicbrainz:{mbid}"
        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
        if cached_hash is not None:
            return cached_hash
        content = await musicbrainz_client.fetch_cover_art(mbid)
        if content is None:
            return None
        thumb_hash = blake3.blake3(content).hexdigest()
        await media_cache.put_thumb(thumb_hash, synthetic_id, content)
        return thumb_hash

    async def _do_audio_transcode_request(self, msg: dict) -> None:
        """
        docs/musicbay.md's one exception to "no node-side transcode pool":
        WMA and Musepack tag/cover fine (enrich_audio.py) but decode in no
        mainstream browser's <audio> element at all. Transcoded to AAC/M4A
        once and cached under its own content hash — same "computed once,
        reused forever" shape as `_fetch_and_cache_poster`/`_cover` above,
        served back to the client through the ordinary file_req/chunk path
        (`_try_serve_thumbnail`, generalized to multi-chunk for this) rather
        than a new download mechanism.
        """
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return
        file_path = entry_abs_path(ctx["roots"], entry)
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        media_cache = self._ctx.get("media_cache")
        if media_cache is None:
            self._send({"type": "error", "detail": "Transcoding unavailable"})
            return

        synthetic_id = f"audio_transcode:{entry.id}"
        cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
        if cached_hash is not None:
            blob = await media_cache.get_thumb(cached_hash)
            if blob is not None:
                self._send({"type": MNP.AUDIO_TRANSCODE_RESP, "v": MNP_VERSION,
                           "file_id": file_id, "hash": cached_hash,
                           "size": len(blob), "mime": "audio/mp4"})
                return
            # Cached hash but the blob itself was pruned: fall through and
            # transcode again below, same as a cold cache.

        sem = self._transcode_semaphore()
        if sem.locked() and sem._value <= 0:
            self._send({"type": "error", "detail": "Server busy, retry shortly"})
            return
        async with sem:
            try:
                blob = await _transcode_audio_to_aac(file_path)
            except Exception as e:
                log.warning("Audio transcode failed for %s: %s", entry.id[:12], e)
                self._send({"type": "error", "detail": f"Transcode failed: {e}"})
                return

        transcode_hash = blake3.blake3(blob).hexdigest()
        await media_cache.put_thumb(transcode_hash, synthetic_id, blob)
        self._audit("audio_transcode", entry.name)
        self._send({"type": MNP.AUDIO_TRANSCODE_RESP, "v": MNP_VERSION,
                   "file_id": file_id, "hash": transcode_hash,
                   "size": len(blob), "mime": "audio/mp4"})

    async def _do_music_meta_request(self, msg: dict) -> None:
        """
        docs/musicbay.md §4.3: MusicBrainz metadata for one track, resolved
        from the group's index by its content id. Album-level (release), the
        direct analogue of Videos' show-level TMDB caching: one search per
        (artist, album) pair serves cover art and canonical naming to every
        track of the same release, keyed off the `artist`/`album` fields
        enrich_audio.py already populated at index time (from tags, or the
        filename-parse fallback) — never re-parsed here.

        Keyed by `file_id` (the entry's own content hash), not `path`: found
        live (2026-08-25) — `IndexEntry.path` is the *folder* a file is in
        (indexer.py's `_virtual_dir`), so any two tracks in the same folder
        (routinely true — an album is one folder, many tracks) shared the
        same `.path`, and looking a track up by it silently resolved to
        whichever entry happened to be first in the index. Three unrelated
        albums showed the same wrong cover before this fix, all sharing one
        folder with the track that legitimately matched it.
        """
        file_id = msg.get("file_id")
        log.debug("music_meta_req file_id=%r", file_id)
        if not isinstance(file_id, str) or not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return
        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        media_cache = self._ctx.get("media_cache")
        musicbrainz_client = self._ctx.get("musicbrainz_client")
        # Same silent, no-error degradation as _do_media_meta_request: no
        # client configured, MusicBrainz off for this group, or nothing to
        # search with (no artist/album — an untagged, unparseable file) all
        # look identical to the caller, which already has to handle "no
        # match" as the ordinary case in flat mode.
        if (media_cache is None or musicbrainz_client is None
                or not ctx.get("musicbrainz_enabled", True)
                or not entry.artist or not entry.album):
            self._send({"type": MNP.MUSIC_META_RESP, "v": MNP_VERSION,
                       "file_id": file_id, "confidence": 0})
            return

        mbid = await media_cache.get_file_mbid(entry.id)
        meta = await media_cache.get_mbid_meta(mbid) if mbid else None

        if meta is None:
            result, ratio = await musicbrainz_client.search_release(entry.artist, entry.album)
            if result is None or ratio < 0.6:
                self._send({"type": MNP.MUSIC_META_RESP, "v": MNP_VERSION,
                           "file_id": file_id, "confidence": 0})
                return
            mbid = result.get("id")
            artist_credit = result.get("artist-credit") or []
            meta = {
                "artist": artist_credit[0].get("name") if artist_credit else entry.artist,
                "album": result.get("title"),
                "release_date": result.get("date"),
                "confidence": ratio,
            }
            await media_cache.set_file_mbid(entry.id, mbid)
            await media_cache.set_mbid_meta(mbid, meta)

        cover_thumb_hash = await self._fetch_and_cache_cover(
            media_cache, musicbrainz_client, mbid)
        log.debug("music_meta_req file_id=%r: replying mbid=%s cover=%s",
                  file_id, mbid, cover_thumb_hash)

        self._send({
            "type": MNP.MUSIC_META_RESP, "v": MNP_VERSION, "file_id": file_id,
            "mbid": mbid,
            "artist": meta.get("artist"),
            "album": meta.get("album"),
            "title": entry.display_title,
            "release_date": meta.get("release_date"),
            "cover_thumb_hash": cover_thumb_hash,
            "confidence": meta.get("confidence", 1.0),
        })

    async def _do_media_meta_request(self, msg: dict) -> None:
        """
        docs/mediacenter.md §5.4: TMDB metadata for one file, resolved from
        the group's index by its content id (root+relpath the client already
        knows from index_sync/index_delta identify the entry; its own `id`
        is what actually names one file — never a raw filesystem path off
        the wire).

        Keyed by `file_id`, not `path`: `IndexEntry.path` is the *folder* a
        file is in (indexer.py's `_virtual_dir`), so two files in the same
        folder — any multi-episode season, routinely — shared the same
        `.path`, and a lookup by it could silently resolve to the wrong
        entry (found live via the Music app's identical bug, 2026-08-25).
        """
        file_id = msg.get("file_id")
        log.debug("media_meta_req file_id=%r", file_id)
        if not isinstance(file_id, str) or not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return
        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        media_cache = self._ctx.get("media_cache")
        tmdb_client = self._ctx.get("tmdb_client")
        # Per-group, not node-wide (docs/mediacenter.md §5.5, 2026-08-24):
        # treated exactly like "no client configured" — same silent, no-error
        # degradation, since a member's Videos tab already has to handle "no
        # TMDB match" as the ordinary case.
        if media_cache is None or tmdb_client is None or not ctx.get("tmdb_enabled", True):
            self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION,
                       "file_id": file_id, "confidence": 0})
            return

        # A video the indexer has seen but not yet *enriched* has no
        # display_title (enrich.py always sets one) and season/episode still
        # None — so the movie/show split reads "movie" and would hand its raw
        # filename to TMDB's movie search. During a slow initial scan with a
        # browser on the Videos tab that is a storm of
        # `search/movie?query=<raw filename>` (found live 2026-08-29, an
        # 8-minute scan). While un-enriched we never *search*: we serve a
        # cached match if there is one (§ below), else confidence 0 and the
        # client refetches once the index delta carries the enriched fields.
        enriched = bool(entry.display_title)

        is_show = entry.season is not None and entry.episode is not None
        media_type = "tv" if is_show else "movie"

        cached = await media_cache.get_file_tmdb(entry.id)
        meta = None
        tmdb_id = None
        if cached is not None:
            cached_tmdb_id, cached_media_type = cached
            # Serve the cached match when its kind still agrees with the
            # entry's current classification — OR when the entry is not
            # enriched yet: its season/episode aren't populated, so the
            # movie/show split above is not meaningful, and the cached kind
            # (set when this file WAS enriched) is the reliable one. This is
            # what keeps a restart from re-querying TMDB for everything
            # already resolved: the storm was an un-enriched show episode
            # looking like a "movie" and treating its own valid "tv" match
            # as stale.
            #
            # Once enriched, the strict `cached_media_type == media_type`
            # check still stands: an enrichment fix that reclassifies a
            # folder movie->tv must drop the stale movie-era match and
            # re-resolve (found live — a Specials-folder fix left hundreds
            # of files answering with their wrong-kind match forever).
            if cached_media_type == media_type or not enriched:
                media_type = cached_media_type
                is_show = media_type == "tv"
                tmdb_id = cached_tmdb_id
                meta = await media_cache.get_tmdb_meta(tmdb_id, media_type)

        if meta is None:
            if not enriched:
                self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION,
                           "file_id": file_id, "confidence": 0})
                return
            result, ratio = await self._tmdb_search(tmdb_client, entry, is_show)
            if result is None or ratio < 0.6:
                self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION,
                           "file_id": file_id, "confidence": 0})
                return
            tmdb_id = str(result["id"])
            meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, result)
            await media_cache.set_file_tmdb(entry.id, tmdb_id, media_type)
            await media_cache.set_tmdb_meta(tmdb_id, media_type, meta)

        poster_thumb_hash = await self._fetch_and_cache_poster(
            media_cache, tmdb_client, meta.get("poster_path"))
        backdrop_thumb_hash = await self._fetch_and_cache_poster(
            media_cache, tmdb_client, meta.get("backdrop_path"))
        log.debug("media_meta_req file_id=%r: replying tmdb_id=%s poster=%s backdrop=%s",
                  file_id, tmdb_id, poster_thumb_hash, backdrop_thumb_hash)

        resp = {
            "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id,
            "tmdb_id": tmdb_id, "title": meta.get("title"),
            "original_title": meta.get("original_title"),
            "overview": meta.get("overview"),
            "poster_thumb_hash": poster_thumb_hash,
            "backdrop_thumb_hash": backdrop_thumb_hash,
            "release_date": meta.get("release_date"),
            "first_air_date": meta.get("first_air_date"),
            "genres": meta.get("genres", []),
            "vote_average": meta.get("vote_average"),
            "runtime": meta.get("runtime"),
            "cast": meta.get("cast", []),
            "director": meta.get("director"),
            "confidence": meta.get("confidence", 1.0),
        }
        if is_show:
            resp["season"] = entry.season
            resp["episode"] = entry.episode
        self._send(resp)

    async def _do_season_meta_request(self, msg: dict) -> None:
        """
        Per-season TMDB overview/poster/air_date for a multi-season show —
        found live: `media_meta_resp`'s one static show-level overview does
        not necessarily describe every season alike (a season-3-specific
        promotional summary applied to all three seasons of a show).
        `tmdb_id` is whatever the client's own prior `media_meta_resp`
        already resolved — never re-derived from a path here, so this
        never re-runs a TMDB search of its own.
        """
        tmdb_id = msg.get("tmdb_id")
        season = msg.get("season")
        if not isinstance(tmdb_id, str) or not tmdb_id or not isinstance(season, int):
            self._send({"type": "error", "detail": "Missing tmdb_id or season"})
            return
        media_cache = self._ctx.get("media_cache")
        tmdb_client = self._ctx.get("tmdb_client")
        # Per-group, not node-wide (docs/mediacenter.md §5.5, 2026-08-24) —
        # same silent zero-confidence degradation as "no client configured".
        if (media_cache is None or tmdb_client is None
                or not self._group_ctx().get("tmdb_enabled", True)):
            self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION,
                       "tmdb_id": tmdb_id, "season": season, "confidence": 0})
            return

        details = await media_cache.get_season_meta(tmdb_id, season)
        if details is None:
            fetched = await tmdb_client.tv_season(tmdb_id, season)
            if fetched is None:
                self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION,
                           "tmdb_id": tmdb_id, "season": season, "confidence": 0})
                return
            # Same per-field English fallback as _tmdb_build_meta: TMDB
            # returns "" for an untranslated field rather than falling back
            # itself.
            if not fetched.get("overview"):
                fallback = await tmdb_client.tv_season(tmdb_id, season, language="en-US") or {}
                fetched = {**fallback, **{k: v for k, v in fetched.items() if v not in (None, "", [])}}
            await media_cache.set_season_meta(tmdb_id, season, fetched)
            details = fetched

        poster_thumb_hash = await self._fetch_and_cache_poster(
            media_cache, tmdb_client, details.get("poster_path"))
        self._send({
            "type": MNP.SEASON_META_RESP, "v": MNP_VERSION,
            "tmdb_id": tmdb_id, "season": season, "confidence": 1.0,
            "name": details.get("name"),
            "overview": details.get("overview"),
            "air_date": details.get("air_date"),
            "poster_thumb_hash": poster_thumb_hash,
        })

    async def _do_tmdb_search_request(self, msg: dict) -> None:
        """
        Candidate TMDB matches for an operator correcting a wrong automatic
        match (docs/mediacenter.md, §V-whatever this becomes) — a plain
        lookup, not a mutation, so unlike `tmdb_override` this needs no
        admin authority: any member can see what TMDB itself would offer,
        the same as the automatic search already silently does on their
        behalf. Only `tmdb_override` actually changes what everyone sees.
        """
        query = msg.get("query")
        media_type = msg.get("media_type")
        if not isinstance(query, str) or not query.strip() or media_type not in ("movie", "tv"):
            self._send({"type": "error", "detail": "Missing query or media_type"})
            return
        media_cache = self._ctx.get("media_cache")
        tmdb_client = self._ctx.get("tmdb_client")
        # Per-group, not node-wide (docs/mediacenter.md §5.5, 2026-08-24) —
        # same silent empty-results degradation as "no client configured":
        # a member with TMDB off for this group sees the same "type it in
        # yourself" affordance either way, never an error.
        if (media_cache is None or tmdb_client is None
                or not self._group_ctx().get("tmdb_enabled", True)):
            self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION,
                       "query": query, "media_type": media_type, "results": []})
            return

        raw = (await tmdb_client.search_movie_results(query) if media_type == "movie"
              else await tmdb_client.search_tv_results(query))
        results = []
        for r in raw:
            poster_thumb_hash = await self._fetch_and_cache_poster(
                media_cache, tmdb_client, r.get("poster_path"))
            results.append({
                "tmdb_id": str(r.get("id")),
                "title": r.get("title") or r.get("name"),
                "year": (r.get("release_date") or r.get("first_air_date") or "")[:4],
                "poster_thumb_hash": poster_thumb_hash,
            })
        # media_type echoed back, not just query: a client can fire a "war"
        # tv search and a "war" movie search close together, and without it
        # the two responses are indistinguishable for keyed matching
        # (transport.js's tmdb_search_resp handler).
        self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION,
                   "query": query, "media_type": media_type, "results": results})

    def _do_tmdb_override(self, msg: dict) -> None:
        """
        An operator correcting a wrong automatic TMDB match. Signed like
        video_root/tmdb_config: it replaces what every member sees for a
        show/movie, node-wide (media_cache is shared, not per-viewer).

        For a **show**, applied to every entry sharing the representative
        file's display_title — the same grouping the poster grid uses
        (§3.4/§V6) — so the correction sticks regardless of which episode a
        future render picks as representative. For a **movie** it is applied
        to that one file only: guessit gives a whole franchise the same
        display_title, and a fan-out there corrected the wrong films (found
        live, 2026-08-29). See `_admin_exec_tmdb_override`.

        Keyed by `file_id`, not `path` — see `_do_media_meta_request`'s
        docstring for why a folder-level path cannot name one file.
        """
        file_id = msg.get("file_id")
        tmdb_id = msg.get("tmdb_id")
        media_type = msg.get("media_type")
        if not isinstance(file_id, str) or not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return
        if not isinstance(tmdb_id, str) or not tmdb_id or media_type not in ("movie", "tv"):
            self._send({"type": "error", "detail": "Missing tmdb_id or media_type"})
            return
        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        subject = f"file_id={file_id},tmdb_id={tmdb_id},media_type={media_type}"
        self._issue_admin_challenge(OP_TMDB_OVERRIDE, subject)

    async def _admin_exec_tmdb_override(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        subject = pending["subject"]
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"tmdb_override:{subject}")
            return
        fields = dict(part.split("=", 1) for part in subject.split(","))
        file_id, tmdb_id, media_type = fields["file_id"], fields["tmdb_id"], fields["media_type"]

        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        media_cache = self._ctx.get("media_cache")
        if entry is None or media_cache is None:
            self._send({"type": "error", "detail": "File or media cache not available"})
            return
        # This only ever recorded the file->tmdb_id mapping, never the
        # metadata tmdb_id names — _do_media_meta_request's cache check
        # (entry, cache) both agree on media_type, so it trusted the
        # mapping — but found nothing under this *new* id in tmdb_meta
        # (nothing had ever fetched it), and silently fell through to a
        # fresh search using the file's own title, exactly the one that
        # produced the wrong match in the first place. Confirmed live: an
        # override "stuck" for shows only because their own title happened
        # to be enough for that fallback search to land on the right
        # answer anyway, coincidentally — never because the override itself
        # was actually being honored — and was invisible until a movie
        # whose own title search kept landing on the same wrong result
        # exposed it. Fetching and storing the real metadata up front is
        # what makes the *override* the thing a later lookup finds.
        tmdb_client = self._ctx.get("tmdb_client")
        if tmdb_client is not None:
            meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, {})
            await media_cache.set_tmdb_meta(tmdb_id, media_type, meta)
        # A show's episodes are many files that legitimately share one match,
        # and which episode a render picks as representative rotates — so a
        # show override fans out across every entry with the same
        # display_title. A *movie* is one file: fanning out by display_title
        # there is a bug — guessit gives every
        # "<franchise> - <year> - <subtitle>.mkv" the same display_title, so
        # "Fix match" on one entry rewrote the whole franchise (found live,
        # 2026-08-29). Each corrected file is also marked as a manual
        # override so ops.rematch_video / a rename never wipe it.
        is_show = entry.season is not None and entry.episode is not None
        if is_show:
            target_title = entry.display_title or entry.name
            matched = [e for e in ctx["index"].entries
                      if e.type == "video" and (e.display_title or e.name) == target_title]
        else:
            matched = [entry]
        for e in matched:
            await media_cache.set_file_tmdb(e.id, tmdb_id, media_type)
            await media_cache.mark_tmdb_override(e.id)
        self._audit("tmdb_override", subject)

        notice = {"type": MNP.TMDB_OVERRIDE_ACK, "v": MNP_VERSION, "file_id": file_id,
                 "tmdb_id": tmdb_id, "media_type": media_type}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    def _do_tmdb_rematch(self, msg: dict) -> None:
        """
        An operator dropping one file's cached TMDB match so it re-resolves
        with the current matcher (§10.1/V13) — the one-click alternative to
        the full search-and-pick "Fix match" flow, and reachable without
        SSH (`meshbay-node video rematch` clears a whole group). Signed like
        `tmdb_override`: `media_cache` is shared node-wide.
        """
        file_id = msg.get("file_id")
        if not isinstance(file_id, str) or not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return
        if not self._group_ctx()["index"].get_entry(file_id):
            self._send({"type": "error", "detail": "File not found"})
            return
        if not self._has_admin_authority():
            self._send({"type": "error", "detail": "No authorized key for this"})
            return
        self._issue_admin_challenge(OP_TMDB_REMATCH, f"file_id={file_id}")

    async def _admin_exec_tmdb_rematch(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        subject = pending["subject"]
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"tmdb_rematch:{subject}")
            return
        file_id = dict(part.split("=", 1) for part in subject.split(","))["file_id"]
        media_cache = self._ctx.get("media_cache")
        if media_cache is None:
            self._send({"type": "error", "detail": "Media cache not available"})
            return
        await media_cache.drop_tmdb_match(file_id)
        self._audit("tmdb_rematch", subject)

        notice = {"type": MNP.TMDB_REMATCH_ACK, "v": MNP_VERSION, "file_id": file_id}
        for uid, session in list(self._peer_registry().items()):
            try:
                session._send(notice)
            except Exception:
                pass

    async def _tmdb_search(self, tmdb_client, entry, is_show: bool):
        """
        §3.3's retry ladder — same shape for movies and shows (§10.1/V8).
        TMDB's own top result is still trusted per query (§3.3's last row —
        no local re-ranking of *its* list); what the ladder adds is that it
        *scores every candidate query* and keeps the best, instead of
        returning the first that merely clears 0.6.

        The bare parsed title is the weakest query: guessit drops a
        "Volume 2", strips a real subtitle into `alternative_title`, renders
        a sequel index where TMDB spells it differently, and a show's folder
        name can carry a year or release-group noise. A wrong entry that
        scored ~0.7 against that weak query — a same-year making-of
        documentary, a franchise entry whose localized TMDB title *is* the
        franchise name, a season-specific promo entry standing in for a
        whole show — used to win outright before a stronger candidate was
        ever tried. Found live (2026-08-29).
        """
        from meshbay_node.indexer import title_parse

        if is_show:
            title = entry.display_title or title_parse.naive_title(entry.name)
            name_naive = title_parse.naive_title(entry.name)
            year = title_parse.year_in(title) or title_parse.year_in(entry.name)
            extra = [c for c in (name_naive, title_parse.clean_query(title))
                     if c and c != title]
            return await self._tmdb_ladder(
                tmdb_client.search_tv, title, extra, year, strong_extra=False)

        parsed = title_parse.parse_movie_filename(entry.name)
        title = entry.display_title or parsed.display_title or parsed.naive_title
        strong = [c for c in (parsed.alt_title, *title_parse.sequel_variants(title)) if c]
        extra = [c for c in (*strong, parsed.naive_title) if c and c != title]
        return await self._tmdb_ladder(
            tmdb_client.search_movie, title, extra, parsed.year,
            strong_extra=bool(strong))

    @staticmethod
    async def _tmdb_ladder(search_fn, primary: str, extra: list[str],
                           year: int | None, strong_extra: bool):
        """
        `search_fn(query, year) -> (result|None, ratio)`. Try `primary`,
        return at once on a confident hit (ratio >= 0.85 — the common case,
        one request). Otherwise score each `extra` candidate and keep the
        best. `strong_extra` says whether `extra` contains anything more
        specific than a punctuation-normalised restatement of `primary`
        (an alternative_title, a sequel variant); when it does not and the
        primary hit is already decent, the remaining calls are skipped
        (§10.1/V11 — they almost never win and cost a round trip each).
        """
        def _year_of(res: dict) -> int | None:
            d = str(res.get("release_date") or res.get("first_air_date") or "")
            return int(d[:4]) if d[:4].isdigit() else None

        def _rescue(res: dict, r: float) -> float:
            # A sub-0.6 hit whose result lands on the exact requested year:
            # TMDB already year-filtered the search, so this is a hard
            # corroboration that the low ratio is a localised/rearranged
            # title, not a wrong entry. Never overrides a confident hit.
            if r < 0.6 and year and _year_of(res) == year:
                return max(r, 0.6)
            return r

        result, ratio = await search_fn(primary, year)
        if result is not None and ratio >= 0.85:
            return result, ratio
        best_result, best_score = (result, ratio) if result is not None else (None, 0.0)
        if best_result is not None:
            best_score = _rescue(best_result, best_score)
        if best_score >= 0.6 and not strong_extra:
            return best_result, best_score

        for candidate in extra:
            if candidate == primary:
                continue
            r2, ratio2 = await search_fn(candidate, year)
            if r2 is None and year:
                # A year-filtered search that finds nothing: the year tag
                # may be an edition/regional year TMDB doesn't carry. Retry
                # the candidate unconstrained before dropping it.
                r2, ratio2 = await search_fn(candidate, None)
            if r2 is None:
                continue
            score2 = _rescue(r2, ratio2)
            if score2 > best_score:
                best_result, best_score = r2, score2
            if best_score >= 0.85:
                break
        return best_result, best_score

    @staticmethod
    async def _tmdb_build_meta(tmdb_client, tmdb_id: str, media_type: str, result: dict) -> dict:
        """
        `result` (the search hit) only carries `genre_ids` and no `runtime`
        at all — the full details endpoint is the actual source for those,
        falling back to the search result for anything details somehow
        lacks (never expected in practice, just avoids a KeyError-shaped
        surprise if TMDB's response ever varies).
        """
        details = (await tmdb_client.tv_details(tmdb_id) if media_type == "tv"
                  else await tmdb_client.movie_details(tmdb_id)) or result
        # TMDB doesn't fall back server-side for a field with no translation
        # in the configured language — it returns "" (or an empty list) for
        # it, not the English text (confirmed live: a French query left
        # `overview` empty for a title TMDB has no French translation for).
        # The TMDB website covers exactly this gap client-side, by falling
        # back to English per field rather than discarding an otherwise-good
        # localized response over one empty one — mirrored here the same
        # way, at field granularity, not by abandoning the whole response.
        if not details.get("overview") or not details.get("poster_path") or not details.get("genres"):
            fallback = (await tmdb_client.tv_details(tmdb_id, language="en-US") if media_type == "tv"
                       else await tmdb_client.movie_details(tmdb_id, language="en-US")) or {}
            details = {**fallback, **{k: v for k, v in details.items() if v not in (None, "", [])}}
        credits = (await tmdb_client.tv_credits(tmdb_id) if media_type == "tv"
                  else await tmdb_client.movie_credits(tmdb_id))
        cast = [{"name": c.get("name"), "character": c.get("character")}
               for c in (credits or {}).get("cast", [])[:10]]
        director = None
        if media_type == "movie":
            director = next(
                (c.get("name") for c in (credits or {}).get("crew", [])
                 if c.get("job") == "Director"), None)
        else:
            # A series has no single director, and `tv_credits`' crew is the
            # aggregate one — routinely empty, and never a "Director" job.
            # TMDB models the equivalent credit as `created_by` on the show
            # itself, which is what its own page shows; several creators are
            # ordinary, and they read as one line in the detail modal.
            director = ", ".join(
                name for name in
                (c.get("name") for c in details.get("created_by") or [])
                if name) or None
        runtime = details.get("runtime")
        if runtime is None and media_type == "tv":
            episode_run_times = details.get("episode_run_time") or []
            runtime = episode_run_times[0] if episode_run_times else None
        return {
            "title": details.get("title") or details.get("name"),
            "original_title": details.get("original_title") or details.get("original_name"),
            "overview": details.get("overview"),
            "poster_path": details.get("poster_path"),
            "backdrop_path": details.get("backdrop_path"),
            "release_date": details.get("release_date"),
            "first_air_date": details.get("first_air_date"),
            "genres": [g.get("name") for g in details.get("genres", []) if g.get("name")],
            "vote_average": details.get("vote_average"),
            "runtime": runtime,
            "cast": cast,
            "director": director,
        }

    def _do_stream_segment(self, msg: dict) -> None:
        self._spawn(self._do_stream_segment_async(msg))

    async def _do_stream_segment_async(self, msg: dict) -> None:
        """
        Legacy HLS segment extraction (superseded by stream_req/MSE).

        Finding H6: this ran subprocess.run(..., timeout=30) directly inside the
        event loop, so a single request stalled the whole daemon — every peer,
        every group — for up to thirty seconds. Now async and under the same
        transcode semaphore as _stream_video.
        """
        ctx = self._group_ctx()
        file_id = msg["file_id"]
        segment_index = msg["segment_index"]
        segment_duration = msg.get("segment_duration", 4)

        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path = entry_abs_path(ctx["roots"], entry)
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        sem = self._transcode_semaphore()

        try:
            async with sem:
                proc = await asyncio.create_subprocess_exec(
                    platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error",
                    "-ss", str(segment_index * segment_duration),
                    "-i", str(file_path),
                    "-t", str(segment_duration),
                    "-c:v", "copy", "-c:a", "copy",
                    "-f", "mpegts", "pipe:1",
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.DEVNULL,
                )
                try:
                    stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30)
                except asyncio.TimeoutError:
                    proc.kill()
                    await proc.wait()
                    self._send({"type": "error", "detail": "Segment extraction timed out"})
                    return
            if proc.returncode != 0 or not stdout:
                self._send({"type": "error", "detail": "Segment extraction failed"})
                return
            segment_data = stdout
        except Exception:
            self._send({"type": "error", "detail": "Segment extraction failed"})
            return

        self._send({
            "type": MNP.STREAM_SEGMENT,
            "v": MNP_VERSION,
            "file_id": file_id,
            "segment_index": segment_index,
            "data_b64": base64.b64encode(segment_data).decode(),
            "size": len(segment_data),
        })

    def _do_chat_message(self, msg: dict) -> None:
        # Per-group store — see _peer_registry() and finding H1. Reading chat_store
        # off the shared transport context sent every group's messages to the first
        # group's database, and served them back to anyone on the node.
        chat_store = self._group_ctx().get("chat_store")
        payload = msg.get("payload", "")
        sender_name = msg.get("sender_name", "")
        if sender_name:
            self._user_names()[self._user_id] = sender_name
        if chat_store:
            raw = payload.encode() if isinstance(payload, str) else payload
            self._spawn(chat_store.save_message(
                sender_id=self._user_id,
                iteration=msg.get("iteration", 0),
                payload=raw,
                thread_id=msg.get("thread_id"),
                sender_name=sender_name,
            ))

        peers = self._peer_registry()
        broadcast = {
            "type": MNP.CHAT_MESSAGE,
            "v": MNP_VERSION,
            "sender_id": self._user_id,
            "sender_name": sender_name,
            "payload": payload,
            "thread_id": msg.get("thread_id"),
            "timestamp": __import__("time").time(),
        }
        for uid, session in list(peers.items()):
            if uid != self._user_id and session is not self:
                try:
                    session._send(broadcast)
                except Exception:
                    pass

        hub_ws = self._ctx.get("hub_ws")
        if hub_ws and self._group_id:
            try:
                import json as _json
                self._spawn(hub_ws.send(_json.dumps({
                    "type": "chat_notify",
                    "group_id": self._group_id,
                    "sender_name": sender_name,
                    # Who actually wrote it, from the authenticated session. The
                    # hub used to fall back to this node's own token subject —
                    # the operator — so everyone was notified of their own
                    # messages and the operator was notified of nobody's.
                    "sender_user_id": self._user_id,
                })))
            except Exception:
                pass

        self._send({"type": "ack", "v": MNP_VERSION})
        self._audit("chat_message")

    def _do_ping(self, msg: dict) -> None:
        """Answer a liveness probe on an open channel, echoing the caller's token.

        Echoed rather than bare so a client can match the answer to the probe it
        sent and measure a round trip, instead of being reassured by a reply to
        some earlier one.
        """
        self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")})

    def _do_chat_history(self, msg: dict) -> None:
        chat_store = self._group_ctx().get("chat_store")
        if not chat_store:
            self._send({
                "type": MNP.CHAT_HISTORY_RESPONSE,
                "v": MNP_VERSION,
                "messages": [],
                "has_more": False,
            })
            return

        # `before` pages backwards from the newest, which is the direction a chat
        # is actually read. `since` remains for callers that want everything
        # after a point in time; the browser no longer uses it.
        before = msg.get("before")
        limit = max(1, min(int(msg.get("limit", 100)), 200))
        self._spawn(self._send_chat_history(chat_store, before, limit))

    async def _send_chat_history(self, chat_store, before, limit: int) -> None:
        if before:
            msgs = await chat_store.get_before(int(before), limit=limit)
        else:
            msgs = await chat_store.get_recent(limit=limit)
        # Whether the "load older" control has anything left to fetch. Asked
        # about the oldest row returned, so an empty page correctly says no.
        has_more = await chat_store.has_before(msgs[0].id) if msgs else False
        names = self._user_names()
        self._send({
            "type": MNP.CHAT_HISTORY_RESPONSE,
            "v": MNP_VERSION,
            "has_more": has_more,
            "messages": [
                {
                    "id": m.id,
                    "sender_id": m.sender_id,
                    "sender_name": m.sender_name or names.get(m.sender_id, ""),
                    "payload": m.payload.decode("utf-8", errors="replace")
                    if isinstance(m.payload, bytes) else m.payload,
                    "timestamp": m.timestamp,
                    "thread_id": m.thread_id,
                }
                for m in msgs
            ],
        })

    def _link_preview_rate_ok(self) -> bool:
        """
        True when this preview fetch is within both the per-connection and the
        node-wide window; records it when so, and both counts are trimmed to the
        window on every call so the lists cannot grow without bound.
        """
        now = time.monotonic()
        w = _LINK_PREVIEW_RATE_WINDOW
        mine = [t for t in getattr(self, "_link_preview_hits", []) if now - t < w]
        node = [t for t in self._ctx.get("link_preview_hits", []) if now - t < w]
        if (len(mine) >= _LINK_PREVIEW_RATE_PER_CONN
                or len(node) >= _LINK_PREVIEW_RATE_NODE):
            self._link_preview_hits = mine
            self._ctx["link_preview_hits"] = node
            return False
        mine.append(now)
        node.append(now)
        self._link_preview_hits = mine
        self._ctx["link_preview_hits"] = node
        return True

    async def _do_link_preview_request(self, msg: dict) -> None:
        """
        Unfurl a URL a member pasted into chat (draft-v6 §2.7 enrichment rule:
        the client asks, the node produces on demand, the asking device
        caches — nothing durable here).

        `linkpreview.safe_url` is the SSRF gate: the URL a *member* chose
        decides an outbound request from the operator's machine, so http(s)
        only and the resolved address must be globally routable. Failure of
        any kind — blocked, unreachable, not HTML, nothing worth showing —
        comes back as `ok: false`, the way a TMDB miss does; the client then
        just shows the bare link.
        """
        url = msg.get("url")
        key = url if isinstance(url, str) else ""
        cached = _link_preview_cache_get(key)
        if cached is not None:
            self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION})
            return

        if not self._link_preview_rate_ok():
            # Same shape as any other miss — the client shows the bare link. A
            # rate-limited result is not cached, so it is retried once the
            # window clears rather than pinned as "no preview".
            log.debug("link_preview_req: rate-limited (peer=%s)", self._peer_id)
            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
                        "url": key, "ok": False})
            return

        resp: dict = {"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
                      "url": key, "ok": False}
        try:
            meta = await linkpreview.fetch_preview(url)
            if meta is not None:
                resp.update(ok=True, title=meta["title"],
                            description=meta["description"],
                            site_name=meta["site_name"])
                image_url = meta.get("image_url")
                media_cache = self._ctx.get("media_cache")
                if image_url and media_cache is not None:
                    synthetic_id = f"linkpreview:{image_url}"
                    thumb_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id)
                    if thumb_hash is None:
                        jpeg = await linkpreview.fetch_image(image_url)
                        if jpeg:
                            thumb_hash = blake3.blake3(jpeg).hexdigest()
                            await media_cache.put_thumb(thumb_hash, synthetic_id, jpeg)
                    if thumb_hash:
                        resp["image_thumb_hash"] = thumb_hash
        except Exception as e:
            log.debug("link_preview_req %s: %s", key[:80], e)

        _link_preview_cache_put(key, {k: v for k, v in resp.items()
                                      if k not in ("type", "v")})
        self._send(resp)

    def _do_file_upload(self, msg: dict) -> None:
        ctx = self._group_ctx()
        filename = msg.get("filename", "")
        chunk_index = msg.get("chunk_index", 0)
        total_chunks = msg.get("total_chunks", 1)
        data = msg.get("data")

        if not filename or data is None:
            self._send({"type": "error", "detail": "Missing filename or data",
                        "filename": filename})
            return

        if not SAFE_UPLOAD_NAME.match(filename):
            self._send({"type": "error", "detail": "Invalid filename",
                        "filename": filename})
            return

        roots: RootSet | None = ctx.get("roots")
        if not roots:
            self._send({"type": "error",
                        "detail": "No directories configured for this group",
                        "filename": filename})
            return

        # The client names the root it is uploading into — it is browsing one,
        # and with several writable roots any other choice is a guess. It names
        # a root, never a path: the destination inside it is decided below and
        # is not negotiable, which is what keeps C5a closed.
        #
        # An unknown name is refused rather than falling back to a writable
        # root, because "the file went somewhere else" is discovered weeks
        # later — the same reason the old single upload root was never guessed.
        # A client that names nothing is an MNP 1.0 one, and there was exactly
        # one destination in its world: the first writable root.
        target_root_name = str(msg.get("root") or "").strip()
        upload_root = None
        if target_root_name:
            upload_root = roots.by_name(target_root_name)
            if upload_root is None:
                self._send({"type": "error",
                            "detail": f"No directory named "
                                      f"{target_root_name!r} in this group",
                            "code": "no_such_root",
                            "filename": filename})
                return
        else:
            writable = roots.writable_roots
            upload_root = writable[0] if writable else None

        if upload_root is None:
            self._send({"type": "error",
                        "detail": "No writable directory in this group",
                        "code": "no_writable_root",
                        "filename": filename})
            return
        if not upload_root.writable:
            self._send({"type": "error",
                        "detail": f"Directory '{upload_root.name}' is read-only",
                        "code": "root_read_only",
                        "filename": filename})
            self._audit("upload_refused", filename[:64])
            return
        if not upload_root.available:
            self._send({"type": "error",
                        "detail": f"Directory '{upload_root.name}' is "
                                  f"currently unavailable",
                        "code": "root_unavailable",
                        "filename": filename})
            return

        if upload_root.direct:
            rel_dir = upload_root.name
            target_dir = upload_root.path
        else:
            rel_dir = f"{upload_root.name}/{UPLOAD_DIR_NAME}"
            target_dir = upload_root.path / UPLOAD_DIR_NAME
            try:
                target_dir.mkdir(parents=True, exist_ok=True)
            except OSError as e:
                log.warning("Cannot create upload folder in root %r: %s",
                            upload_root.name, e)
                self._send({"type": "error", "detail": "Upload folder unavailable",
                            "filename": filename})
                return

        upload_key = f"{rel_dir}/{filename}"
        state = self._uploads.get(upload_key)
        # A shared directory means two people can send the same name. Refusing the
        # second is safe but silly — everyone's camera produces IMG_1234.jpg — so
        # a free name is found instead. Never a replacement.
        stored_name = state["stored_name"] if state else _free_name(target_dir, filename)
        tmp_path = target_dir / f"{stored_name}.part"
        final_path = target_dir / stored_name

        if chunk_index == 0:
            # Backstop: _free_name already guarantees this, and it stays because
            # it asserts the invariant where the write happens.
            if final_path.exists():
                self._send({"type": "error", "detail": "File already exists",
                        "filename": filename})
                return
            state = {"next_index": 0, "bytes": 0, "stored_name": stored_name}
            self._uploads[upload_key] = state
        elif state is None:
            self._send({"type": "error", "detail": "Upload not started",
                        "filename": filename})
            return

        # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
        # blindly to whatever .part file is already on disk.
        if chunk_index != state["next_index"]:
            self._send({"type": "error", "detail": "Unexpected chunk index",
                        "filename": filename})
            return

        if isinstance(data, str):
            chunk_bytes = base64.b64decode(data)
        else:
            chunk_bytes = bytes(data)

        if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES:
            self._uploads.pop(upload_key, None)
            tmp_path.unlink(missing_ok=True)
            self._send({"type": "error", "detail": "Upload exceeds size limit",
                        "filename": filename})
            return

        with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f:
            f.write(chunk_bytes)
        state["next_index"] = chunk_index + 1
        state["bytes"] += len(chunk_bytes)

        self._send({
            "type": MNP.FILE_UPLOAD_ACK,
            "v": MNP_VERSION,
            "chunk_index": chunk_index,
            "filename": filename,
            # What it is actually called on disk, which a chat attachment has to
            # reference and the uploader deserves to be told.
            "stored_as": stored_name,
            "dir": rel_dir,
        })

        if chunk_index + 1 >= total_chunks:
            self._uploads.pop(upload_key, None)
            tmp_path.rename(final_path)
            log.info("Upload complete: %s (%d chunks, %d bytes)",
                     stored_name, total_chunks, state["bytes"])
            self._audit("file_upload", f"{rel_dir}/{stored_name}")
            self._register_uploader(ctx, rel_dir, stored_name)

    def _register_uploader(self, ctx: dict, rel_dir: str, filename: str) -> None:
        """
        Tag the index entry with the uploader's identity after upload completes.

        The key recorded here is the one this node pinned, not the one the token
        carried. `pk_user` was a hub-chosen claim, and it decided who could later
        delete the file: a hub issuing a token naming its own key could delete
        anyone's uploads on any node. Deletion is supposed to be authorized by the
        node, and this closes the last place where it was not.
        """
        idx = ctx.get("index")
        if not idx:
            return
        for entry in idx.entries:
            if entry.name == filename and entry.path == rel_dir:
                entry.uploader_id = self._user_id
                entry.uploader_pk = self._pinned_pk
                return

    def _do_file_delete(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        if not file_id:
            self._send({"type": "error", "detail": "Missing file_id"})
            return

        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        has_uploader_pk = bool(entry.uploader_pk)
        if not self._has_admin_authority() and not has_uploader_pk:
            self._send({"type": "error", "detail": "No authorized key for deletion"})
            return

        self._issue_admin_challenge(OP_FILE_DELETE, file_id)

    # ── Admin operation challenge/response (finding H5) ──────────────────────

    def _node_pk_b64(self) -> str:
        return pk_to_b64(self._ctx["sk_node"].public_key())

    def _issue_admin_challenge(
        self, op: str, subject: str, payload: dict | None = None,
        group_id: str | None = None,
    ) -> None:
        """
        Ask the client to authorize `op` on `subject` with its Ed25519 identity key.

        The client is sent the transcript *fields*, not opaque bytes, so it can
        rebuild and inspect what it signs. The node keeps the authoritative copy and
        rebuilds the transcript itself at verification time — nothing signed is ever
        taken from the response message.

        `group_id` overrides the connection's group for cross-group operations
        (e.g. root management from a NodePage connection).
        """
        gid = group_id if group_id is not None else (self._group_id or "")
        nonce = os.urandom(32)
        ts = int(time.time())
        op_id = base64.b64encode(os.urandom(16)).decode()
        self._admin_ops[op_id] = {
            "op": op, "subject": subject, "nonce": nonce, "ts": ts,
            "payload": payload or {}, "group_id": gid,
        }
        self._send({
            "type": MNP.ADMIN_CHALLENGE,
            "v": MNP_VERSION,
            "op_id": op_id,
            "op": op,
            "subject": subject,
            "nonce": base64.b64encode(nonce).decode(),
            "ts": ts,
            "node_pk": self._node_pk_b64(),
            "group_id": gid,
        })

    @staticmethod
    def _verify_sig(pk: Ed25519PublicKey | None, transcript: bytes, sig: bytes) -> bool:
        if pk is None:
            return False
        try:
            pk.verify(sig, transcript)
            return True
        except Exception:
            return False

    async def _load_pinned_pk(self) -> None:
        """Remember which key this node pinned for the peer we just authenticated."""
        roster = self._ctx.get("roster")
        if roster is None or not self._user_id:
            return
        ident = await roster.get_identity(self._user_id)
        if ident:
            self._pinned_pk = ident["pk_ed25519"]

    def _is_node_admin(self) -> bool:
        """
        Whether the peer on this connection is the node's operator.

        Was written out twice — once in the handshake ack and once at the gate
        below it — which is how the two come to disagree. From the node's own
        record of who it belongs to, never from a hub claim.
        """
        node_user_id = self._ctx.get("node_user_id")
        return bool(node_user_id and self._user_id == node_user_id)

    def _has_admin_authority(self) -> bool:
        """
        Cheap synchronous pre-check: is there anyone who could authorize this?

        Only decides whether to issue a challenge at all — the gate is
        `_verify_admin_sig`. The flag is set at startup and refreshed in-process
        when an operator pairs.
        """
        return bool(self._ctx.get("has_admin_authority"))

    async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool:
        """
        Check a signature against every key holding node-operator authority.

        Read from the roster on each call rather than cached: revoking a paired
        browser must take effect immediately, and admin operations are rare enough
        that a SQLite read costs nothing.

        There is one source of operator authority and this is it. `admin_pk_ed25519`
        in node.toml used to be honoured alongside the roster; it is gone, and a
        config that still names it is warned about at startup rather than obeyed.
        """
        roster = self._ctx.get("roster")
        if roster is None:
            return False
        for pk_b64 in await roster.operator_pks():
            try:
                pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64))
            except Exception:
                continue
            if self._verify_sig(pk, transcript, sig):
                return True
        return False

    def _do_admin_response(self, msg: dict) -> None:
        op_id = msg.get("op_id", "")
        sig_b64 = msg.get("signature", "")

        pending = self._admin_ops.pop(op_id, None)
        if not pending:
            self._send({"type": "error", "detail": "No pending admin operation"})
            return

        if time.time() - pending["ts"] > ADMIN_CHALLENGE_TTL:
            self._send({"type": "error", "detail": "Admin challenge expired"})
            return

        try:
            sig_bytes = base64.b64decode(sig_b64)
        except Exception:
            self._send({"type": "error", "detail": "Invalid signature encoding"})
            return

        transcript = admin_transcript(
            op=pending["op"],
            node_pk_b64=self._node_pk_b64(),
            group_id=pending["group_id"] if pending.get("group_id") is not None else (self._group_id or ""),
            subject=pending["subject"],
            nonce=pending["nonce"],
            ts=pending["ts"],
        )

        if pending["op"] == OP_FILE_DELETE:
            self._spawn(
                self._admin_exec_file_delete(pending, transcript, sig_bytes))
        elif pending["op"] == OP_DIR_DELETE:
            self._spawn(
                self._admin_exec_dir_delete(pending, transcript, sig_bytes))
        elif pending["op"] == OP_MEMBER_REVOKE:
            self._spawn(
                self._admin_exec_member_revoke(pending, transcript, sig_bytes))
        elif pending["op"] == OP_INVITE_CREATE:
            self._spawn(
                self._admin_exec_invite_create(pending, transcript, sig_bytes))
        elif pending["op"] == OP_GEK_ROTATE:
            self._spawn(
                self._admin_exec_gek_rotate(pending, transcript, sig_bytes))
        elif pending["op"] == OP_MEMBER_UNPIN:
            self._spawn(
                self._admin_exec_member_unpin(pending, transcript, sig_bytes))
        elif pending["op"] == OP_MEMBER_UPLOAD:
            log.warning("Deprecated OP_MEMBER_UPLOAD signed op — use root "
                        "writable/read-only instead")
            self._send({"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
                        "allowed": True, "deprecated": True})
        elif pending["op"] == OP_APPS_ENABLED:
            self._spawn(
                self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
        elif pending["op"] == OP_SET_SCAN_SETTINGS:
            self._spawn(
                self._admin_exec_set_scan_settings(pending, transcript, sig_bytes))
        elif pending["op"] == OP_TMDB_CONFIG:
            self._spawn(
                self._admin_exec_tmdb_config(pending, transcript, sig_bytes))
        elif pending["op"] == OP_TMDB_ENABLED:
            self._spawn(
                self._admin_exec_tmdb_enabled(pending, transcript, sig_bytes))
        elif pending["op"] == OP_VIDEO_ROOT:
            self._spawn(
                self._admin_exec_video_root(pending, transcript, sig_bytes))
        elif pending["op"] == OP_TMDB_OVERRIDE:
            self._spawn(
                self._admin_exec_tmdb_override(pending, transcript, sig_bytes))
        elif pending["op"] == OP_TMDB_REMATCH:
            self._spawn(
                self._admin_exec_tmdb_rematch(pending, transcript, sig_bytes))
        elif pending["op"] == OP_MUSICBRAINZ_ENABLED:
            self._spawn(
                self._admin_exec_musicbrainz_enabled(pending, transcript, sig_bytes))
        elif pending["op"] == OP_AUDIO_ROOT:
            self._spawn(
                self._admin_exec_audio_root(pending, transcript, sig_bytes))
        elif pending["op"] == OP_PHOTO_ROOTS:
            self._spawn(
                self._admin_exec_photo_roots(pending, transcript, sig_bytes))
        elif pending["op"] == OP_ROOT_ADD:
            self._spawn(
                self._admin_exec_root_add(pending, transcript, sig_bytes))
        elif pending["op"] == OP_ROOT_REMOVE:
            self._spawn(
                self._admin_exec_root_remove(pending, transcript, sig_bytes))
        elif pending["op"] == OP_ROOT_UPDATE:
            self._spawn(
                self._admin_exec_root_update(pending, transcript, sig_bytes))
        elif pending["op"] == OP_ROOT_EJECT:
            self._spawn(
                self._admin_exec_root_eject(pending, transcript, sig_bytes))
        elif pending["op"] == OP_ROOT_PLUG:
            self._spawn(
                self._admin_exec_root_plug(pending, transcript, sig_bytes))
        elif pending["op"] == OP_GROUP_ATTACH:
            self._spawn(
                self._admin_exec_group_attach(pending, transcript, sig_bytes))
        elif pending["op"] == OP_GROUP_DETACH:
            self._spawn(
                self._admin_exec_group_detach(pending, transcript, sig_bytes))
        else:
            self._send({"type": "error", "detail": "Unknown admin operation"})

    async def _admin_exec_file_delete(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        file_id = pending["subject"]
        ctx = self._group_ctx()
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        uploader_pk = None
        if entry.uploader_pk:
            try:
                uploader_pk = Ed25519PublicKey.from_public_bytes(
                    base64.b64decode(entry.uploader_pk))
            except Exception:
                uploader_pk = None

        # Node operator, or the user who uploaded this file — verified by the key
        # recorded at upload time, never by a JWT claim (the hub controls those).
        if not (await self._verify_admin_sig(transcript, sig)
                or self._verify_sig(uploader_pk, transcript, sig)):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
            return

        self._exec_file_delete(ctx, file_id, entry)

    async def _admin_exec_invite_create(
        self, pending: dict, transcript: bytes, sig: bytes,
    ) -> None:
        # Node operator only. A group admin who does not run the node has no
        # authority over who this node admits (deny by default). Delegation is
        # designed but deferred — see §6.2 of docs/invite-pairing-v1.md.
        if not await self._verify_admin_sig(transcript, sig):
            self._send({"type": "error", "detail": "Signature verification failed"})
            self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}")
            return

        payload = pending["payload"]
        try:
            result = await self._run_op(
                ops.create_invite,
                payload["group_id"],
                payload.get("username", ""),
                user_id=payload["user_id"],
                created_by=self._user_id or "",
            )
        except ops.OpError as e:
            self._send({"type": "error", "detail": e.message})
            return

        self._audit("invite_create", f"target={payload['user_id'][:8]}")
        self._send({
            "type": MNP.INVITE_RESULT,
            "v": MNP_VERSION,
            "code": result["code"],
            "expires_at": result["expires_at"],
            "user_id": result["user_id"],
            "username": result.get("username", ""),
        })

    def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
        file_path = entry_abs_path(ctx["roots"], entry)
        if file_path.exists():
            file_path.unlink()
            log.info("File deleted: %s", entry.name)
            self._audit("file_delete", entry.name)

        ctx["index"].remove_entry(file_id)
        self._send({
            "type": MNP.FILE_DELETE_ACK,
            "v": MNP_VERSION,
            "file_id": file_id,
        })

    def _grant_stream_credit(self, msg: dict) -> None:
        """
        The client has room for more segments.

        `n` of zero is a keepalive, not a no-op: a viewer whose buffer is
        already a minute and a half ahead of the playhead deliberately grants
        nothing, and must still be able to say it is there. Without that, the
        stall timeout below cannot tell a paused film from a closed tab.
        """
        log.debug("stream credit +%s (had %d, sent %d)",
                  msg.get("n"), self._stream_credit, self._stream_segments)
        try:
            n = int(msg.get("n", 1))
        except (TypeError, ValueError):
            n = 1
        if n == 0:
            # The fingerprint of a client that bounds its read-ahead. A client
            # that never sends one is granting credit per append — which is
            # what fills the browser's buffer ceiling and wedges the player.
            self._stream_keepalives += 1
            if self._stream_keepalives == 1:
                log.info("stream: peer is pacing itself (first keepalive at "
                         "%d segments)", self._stream_segments)
        self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT))
        self._stream_heard_at = time.monotonic()
        self._stream_credit_evt.set()

    def _stop_stream(self) -> None:
        """
        The viewer was closed. Stop transcoding and let go of the slot.

        Without this the only thing that ended a stream was the credit timeout,
        so ffmpeg kept running and held one of the node's two transcode slots
        for two minutes after nobody was watching — which is how closing a video
        made the next one answer "server busy".
        """
        self._stream_stopped = True
        self._stream_credit_evt.set()

    async def _await_stream_credit(self) -> bool:
        """
        Block until the client has room. False if it stopped asking.

        Without this the node hands ffmpeg's entire output to the channel as
        fast as it is produced, and the browser holds a four gigabyte film in a
        JavaScript array while MediaSource consumes it a segment at a time.
        """
        # Measured from the last thing the peer said, not from the start of the
        # wait: a viewer that is buffered well ahead sends keepalives and grants
        # nothing for minutes at a time, and that is a watched film, not a
        # stalled one.
        self._stream_heard_at = time.monotonic()
        waiting_since = 0.0
        while self._stream_credit <= 0:
            if waiting_since == 0.0:
                waiting_since = time.monotonic()
                # Debug: a paced viewer runs out of credit between every
                # window, so this is one line per eight segments — hundreds
                # per film. It is worth having, but not by default.
                log.debug("stream: out of credit at %d segments (%.0f MB) — "
                          "waiting for the peer",
                          self._stream_segments,
                          self._stream_segments * STREAM_SEGMENT_SIZE / 1048576)
            if self._stream_stopped:
                return False
            # Checked before the wait as well as after it: a peer that vanishes
            # sends no credit and fires no event, so waiting the full timeout
            # on a channel that is already shut is pure dead time on a slot.
            if self._channel is None or self._channel.readyState != "open":
                return False
            self._stream_credit_evt.clear()
            try:
                # In slices rather than one long sleep, so a connection that
                # dies mid-wait is noticed in seconds instead of minutes. The
                # total budget is unchanged.
                await asyncio.wait_for(self._stream_credit_evt.wait(),
                                       timeout=STREAM_CREDIT_POLL)
            except asyncio.TimeoutError:
                silent = time.monotonic() - self._stream_heard_at
                if silent >= STREAM_CREDIT_TIMEOUT:
                    log.info("Stream stalled: nothing from peer=%s for %.0fs",
                             (self._user_id or "?")[:8], silent)
                    return False
                continue
            if self._stream_stopped:
                return False
            if self._channel is None or self._channel.readyState != "open":
                return False
        if waiting_since:
            waited_for = time.monotonic() - waiting_since
            # Only a wait long enough to be a symptom. Normal pacing puts a
            # gap of a few seconds between windows; a minute means the viewer
            # is buffered right up and playing, or has stopped watching.
            level = log.info if waited_for >= 10 else log.debug
            level("stream: credit arrived after %.1fs", waited_for)
        self._stream_credit -= 1
        return True

    async def _replace_stream(self, msg: dict) -> None:
        """Retire this session's previous stream before starting another.

        A viewer plays one film at a time, so a second request means the first
        one is finished whatever the client managed to tell us. Relying on
        `stream_stop` alone was not enough: a browser that is backgrounded,
        reloaded or simply loses the message never sends it, and the only other
        thing that ends a stream is STREAM_CREDIT_TIMEOUT — two minutes during
        which ffmpeg keeps running and holds one of the node's two transcode
        slots.

        That is the reported failure exactly: first video fine, second fine,
        third answered "Server busy" because the first two were still holding
        both slots. The client shows that as "buffering" forever.

        Waiting for the old task is what makes the slot available: it is the
        exit of its `async with sem` that releases it.
        """
        prev = self._stream_task
        if prev is not None and not prev.done():
            t0 = time.monotonic()
            log.info("stream: retiring previous stream")
            self._stop_stream()
            try:
                await asyncio.wait_for(asyncio.shield(prev), timeout=15)
                log.info("stream: previous stream ended in %.1fs",
                         time.monotonic() - t0)
            except asyncio.TimeoutError:
                log.warning("stream: previous stream STILL RUNNING after 15s")
            except Exception:
                pass   # it failed on its own; the slot is free either way
        self._stream_task = asyncio.current_task()
        await self._stream_video(msg)

    def _transcode_semaphore(self) -> asyncio.Semaphore:
        """The node's stream budget, shared across every peer.

        One ffmpeg per request with no cap lets any member exhaust the node's
        CPU and process table (H6). The semaphore lives on the transport
        context rather than the session so that it counts the node's viewers
        and not one browser's, and it is created once: rebuilding it per call
        would hand every caller its own budget and cap nothing at all.
        """
        sem = self._ctx.get("_transcode_sem")
        if sem is None:
            n = self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES
            sem = asyncio.Semaphore(n)
            self._ctx["_transcode_sem"] = sem
            log.info("stream: %d concurrent viewers allowed", n)
        return sem

    async def _stream_video(self, msg: dict) -> None:
        """Stream a video file as fMP4 segments via MSE-compatible output."""
        sem = self._transcode_semaphore()
        if sem.locked() and sem._value <= 0:
            self._send({"type": "error", "detail": "Server busy, retry shortly"})
            return
        log.info("stream: waiting for a slot (free=%s)", sem._value)
        async with sem:
            log.info("stream: slot acquired (free=%s)", sem._value)
            try:
                await self._stream_video_inner(msg)
            finally:
                log.info("stream: slot released (free=%s)", sem._value + 1)

    async def _stream_video_inner(self, msg: dict) -> None:
        ctx = self._group_ctx()
        file_id = msg.get("file_id", "")
        entry = ctx["index"].get_entry(file_id)
        if not entry:
            self._send({"type": "error", "detail": "File not found"})
            return

        file_path = entry_abs_path(ctx["roots"], entry)
        if not file_path.exists():
            self._send({"type": "error", "detail": "File not on disk"})
            return

        gek = ctx.get("gek")
        file_hash = bytes.fromhex(entry.id)

        try:
            codec_str, duration, has_audio, _width, _height, raw_video_codec = \
                await _probe_video(str(file_path))
        except Exception as e:
            self._send({"type": "error", "detail": f"Probe failed: {e}"})
            return

        if not codec_str:
            self._send({"type": "error", "detail": "Unsupported video codec"})
            return

        # Where to begin. Seeking is a stream restarted somewhere else: the
        # viewer moves the scrubber, this session's previous stream is retired
        # by _replace_stream, and ffmpeg is spawned again with -ss.
        try:
            start = float(msg.get("start", 0) or 0)
        except (TypeError, ValueError):
            start = 0.0
        # Past the end would produce an empty stream and a player waiting for
        # segments that are never coming.
        if duration and start >= duration - 1:
            start = max(0.0, duration - 5)
        start = max(0.0, start)

        # -ss BEFORE -i, which seeks by the container index rather than by
        # decoding up to the point: milliseconds on a 500 MB film instead of
        # tens of seconds. It lands on the keyframe at or before `start`, so
        # the picture can begin a few seconds earlier than asked — which is
        # what every streaming player does, and why the client is told the
        # value used rather than left to assume its own.
        seek_args = ["-ss", f"{start:.3f}"] if start > 0 else []
        # Video is copied whenever the browser can decode it directly —
        # re-encoding it is the expensive thing this pipeline exists to avoid,
        # and H264/VP9/AV1 already decode fine in-browser. HEVC is the one
        # exception (BROWSER_INCOMPATIBLE_VIDEO_CODECS, media_probe.py): found
        # live, a real HEVC/EAC3 WEB-DL reported "Codec not supported for
        # streaming" from MediaSource.isTypeSupported even though ffprobe/VLC
        # play it fine — Chrome has no HEVC decoder on most non-Apple
        # platforms. The operator can turn this fallback off (node.toml
        # transcode_incompatible_video = false) for a client fleet they know
        # already decodes HEVC, since it is real CPU cost, unlike the copy
        # path. Audio is always transcoded to AAC, never copied — see
        # _probe_video for why "copy" there is not an option, not even for a
        # codec that sounds close enough (plain AC-3 has the same in-browser
        # decode problem as E-AC-3, just without ffmpeg also refusing to mux
        # it). Transcoding audio is cheap; it does not change the cost model
        # the transcode-slot semaphore is sized around.
        transcode_video = (
            raw_video_codec in BROWSER_INCOMPATIBLE_VIDEO_CODECS
            and self._ctx.get("transcode_incompatible_video", True)
        )
        map_args = ["-map", "0:v:0"]
        if transcode_video:
            # -pix_fmt yuv420p: a 10-bit or 4:4:4 HEVC source (common for HDR
            # WEB-DLs) fails "-profile:v high" outright otherwise — libx264's
            # High profile is 8-bit 4:2:0 only. Downsampling loses nothing a
            # browser could show anyway (MSE/HTML5 video has no HDR path).
            codec_args = ["-c:v", "libx264", "-pix_fmt", "yuv420p",
                          "-profile:v", "high", "-level", "4.1",
                          "-preset", "veryfast", "-crf", "21"]
            # Must match "-profile:v high -level 4.1" byte-for-byte (avc1.<profile
            # hex><constraint><level hex>) — the client checks this string with
            # MediaSource.isTypeSupported before trusting a single byte of the
            # stream, so a mismatch here fails exactly the check this exists to pass.
            codec_str = "avc1.640029,mp4a.40.2" if has_audio else "avc1.640029"
        else:
            codec_args = ["-c:v", "copy"]
        if has_audio:
            map_args += ["-map", "0:a:0"]
            # Downmixed to stereo: a WEB-DL's 5.1 track becomes 6-channel AAC
            # with no "-ac", which ffprobe and VLC accept fine but which some
            # browsers' MSE decoder rejects outright once real fragments are
            # appended — isTypeSupported() only checks the codec string, so
            # the failure doesn't surface until playback, as a SourceBuffer
            # forced out of its MediaSource with no further explanation.
            codec_args += ["-c:a", "aac", "-ac", "2", "-b:a", "192k"]
        proc = await asyncio.create_subprocess_exec(
            platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error",
            *seek_args,
            "-i", str(file_path),
            *map_args,
            *codec_args,
            "-movflags", "frag_keyframe+empty_moov+default_base_moof",
            "-f", "mp4", "pipe:1",
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )

        self._send({
            "type": MNP.STREAM_INIT,
            "v": MNP_VERSION,
            "file_id": file_id,
            "codec": codec_str,
            "duration": duration,
            # ffmpeg restarts its timestamps at zero whatever we seek to, so
            # this is what the client adds back (`SourceBuffer.timestampOffset`)
            # to put the fragments where they belong on the timeline.
            "start": start,
        })

        # A client that says nothing gets the old behaviour, which is why this
        # defaults to unlimited rather than to zero: a stream that waits for
        # credit from a peer that will never send any is a stream that hangs.
        try:
            self._stream_credit = int(msg.get("credits", 0) or 0)
        except (TypeError, ValueError):
            self._stream_credit = 0
        paced = self._stream_credit > 0
        self._stream_stopped = False

        index = 0
        self._stream_started_at = time.monotonic()
        self._stream_segments = 0
        reason = "eof"
        log.info("stream: stream_init sent file=%s paced=%s credits=%d start=%.1fs",
                 file_id[:12], paced, self._stream_credit, start)
        try:
            while True:
                if paced and not await self._await_stream_credit():
                    reason = "no-credit-or-gone"
                    break
                if self._stream_stopped:
                    reason = "stopped-by-peer"
                    log.info("Stream stopped by peer=%s after %d segments",
                             (self._user_id or "?")[:8], index)
                    break
                data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
                if not data:
                    break
                # Same derivation as a file chunk, indexed by segment: one
                # implementation, in `meshbay_common.protocol`.
                nonce, ct = chunk_ciphertext(gek, data, index, file_hash)
                self._send({
                    "type": MNP.STREAM_DATA,
                    "v": MNP_VERSION,
                    "file_id": file_id,
                    "segment_index": index,
                    "nonce": nonce,
                    "ct": ct,
                    "plaintext_size": len(data),
                })
                index += 1
                self._stream_segments = index
                if index % 100 == 0:
                    # A stream that stops shows up here as a last line, and the
                    # numbers on it say which side stopped it.
                    log.info("stream: %d segments (%.0f MB), credit=%d, "
                             "keepalives=%d, %.0fs in",
                             index, index * STREAM_SEGMENT_SIZE / 1048576,
                             self._stream_credit, self._stream_keepalives,
                             time.monotonic() - self._stream_started_at)
                await asyncio.sleep(0)
        except Exception as e:
            log.error("Stream error: %s", e)
        finally:
            try:
                proc.kill()
            except ProcessLookupError:
                pass
            # `await proc.wait()` on its own is the deadlock the asyncio docs
            # warn about: ffmpeg fills the stdout pipe we have stopped reading,
            # and the transport cannot finish closing until that buffer is
            # drained. Measured on 2026-08-16 with stream: — a viewer closed
            # the player after 99 segments (25 MB) and the task sat here past
            # the 15 s handover timeout, holding a transcode slot. The node has
            # two, so the next video waited and the one after was refused.
            #
            # Drain first, then wait with a bound. The slot must come back even
            # if the process is being stubborn: it has already had SIGKILL, and
            # the OS will reap it whether or not we are still watching.
            stderr_output = b""
            for pipe in (proc.stdout, proc.stderr):
                if pipe is None:
                    continue
                try:
                    drained = await asyncio.wait_for(pipe.read(), timeout=2)
                    if pipe is proc.stderr:
                        stderr_output = drained
                except Exception:
                    pass
            try:
                await asyncio.wait_for(proc.wait(), timeout=5)
            except Exception:
                log.warning("stream: ffmpeg did not reap in 5s — "
                            "releasing the slot regardless")

        # A positive returncode is ffmpeg exiting on its own with an error,
        # before we ever killed it (a kill shows up as a negative signal
        # number instead) — zero segments in that case is a real failure,
        # not a normal end, and saying nothing here is indistinguishable
        # from "the file is just this short". Found live against a real
        # 5.1 E-AC-3 WEB-DL that ffmpeg refused to even start muxing.
        # Detail stays server-side (L3: never hand a peer raw stderr).
        if index == 0 and proc.returncode is not None and proc.returncode > 0:
            log.error("stream: ffmpeg exited rc=%s before any output — %s",
                      proc.returncode,
                      stderr_output.decode(errors="replace").strip().splitlines()[-1:]
                      or "(no stderr)")
            if not self._stream_stopped:
                self._send({"type": "error",
                           "detail": "Could not stream this file"})
        elif not self._stream_stopped:
            self._send({
                "type": MNP.STREAM_END,
                "v": MNP_VERSION,
                "file_id": file_id,
            })
        log.info("stream: stream ended reason=%s segments=%d after %.1fs",
                 reason, index, time.monotonic() - self._stream_started_at)
        log.info("Streamed %s: %d segments", entry.name, index)
        self._audit("stream_video", entry.name)

    def _send(self, obj: dict) -> None:
        if self._channel and self._channel.readyState == "open":
            self._channel.send(_pack(obj))
        else:
            log.warning("WebRTC send skipped: channel=%s",
                        self._channel.readyState if self._channel else "none")

    async def shutdown_tasks(self) -> None:
        """Stop everything this session is doing and give back what it holds.

        Separate from close() because the connection-state handler runs while
        aiortc is already tearing the peer connection down — calling pc.close()
        from in there would re-enter it. What matters for the transcode slot is
        here: cancelling the task runs the exit of its `async with sem`.
        """
        self._stop_stream()
        for task in list(self._tasks):
            task.cancel()
        if self._tasks:
            await asyncio.gather(*self._tasks, return_exceptions=True)

    async def close(self) -> None:
        self._audit("disconnect")
        if self._user_id:
            self._peer_registry().pop(self._user_id, None)
        await self.shutdown_tasks()
        await self._pc.close()


def _read_and_encrypt(
    gek: bytes,
    file_path: Path,
    chunk_index: int,
    file_hash: bytes,
    file_id: str = "",
) -> dict:
    """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short."""
    with open(file_path, "rb") as f:
        f.seek(chunk_index * CHUNK_SIZE)
        plaintext = f.read(CHUNK_SIZE)
    return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id)


async def _transcode_audio_to_aac(file_path: Path) -> bytes:
    """
    One-shot, whole-file transcode to AAC in an M4A container — no live
    piping, no seeking, unlike `_stream_video_inner`'s fMP4 segments: a
    WMA/Musepack source here is a few MB at most, so there is nothing to
    gain from streaming it and a real cost to the added complexity
    (fragmented output needs `-movflags empty_moov` and its own
    client-side reassembly). A plain temp file lets ffmpeg write a normal,
    fully-seekable M4A container instead. `-vn` drops any attached-picture
    "video" stream some taggers embed as cover art — without it, ffmpeg's
    mp4 muxer has been seen treating that picture as a video track to
    encode, which is not what this is for; cover art still comes from the
    ordinary embedded/sibling-file path (enrich_audio.py), never from here.
    """
    fd, tmp_name = tempfile.mkstemp(suffix=".m4a")
    os.close(fd)
    tmp_path = Path(tmp_name)
    try:
        proc = await asyncio.create_subprocess_exec(
            platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", "-y",
            "-i", str(file_path),
            "-vn", "-c:a", "aac", "-ac", "2", "-b:a", "192k",
            "-f", "ipod", str(tmp_path),
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            _, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=AUDIO_TRANSCODE_TIMEOUT_SECS)
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            raise RuntimeError(f"ffmpeg timed out after {AUDIO_TRANSCODE_TIMEOUT_SECS}s")
        if proc.returncode != 0:
            raise RuntimeError(
                f"ffmpeg exited {proc.returncode}: {stderr.decode(errors='replace')[:300]}")
        return tmp_path.read_bytes()
    finally:
        tmp_path.unlink(missing_ok=True)


class WebRTCTransport:
    """
    Manages WebRTC peer connections for browser clients.

    Usage:
        transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, index)
        answer_sdp = await transport.handle_offer(offer_sdp, peer_id)
        # Return answer_sdp to the browser via hub signaling
    """

    def __init__(
        self,
        sk_node: Ed25519PrivateKey,
        hub_pk_pem: bytes,
        gek: bytes,
        roots: RootSet,
        index: GroupIndex,
        groups: dict[str, dict] | None = None,
        denylist: Any | None = None,
        stun_servers: list[str] | None = None,
        max_concurrent_streams: int | None = None,
        transcode_incompatible_video: bool = True,
    ):
        self._ctx: dict[str, Any] = {
            "sk_node": sk_node,
            "hub_pk_pem": hub_pk_pem,
            "gek": gek,
            "roots": roots,
            "index": index,
            "_peers": {},
            # None means "the operator said nothing" — the default applies. It
            # is read once, when the first stream builds the semaphore.
            "max_concurrent_streams": max_concurrent_streams,
            # Operator opt-out (node.toml) for the HEVC-etc. transcode
            # fallback in _stream_video_inner — real CPU cost, unlike copy.
            "transcode_incompatible_video": transcode_incompatible_video,
        }
        if groups:
            self._ctx["groups"] = groups
        if denylist:
            self._ctx["denylist"] = denylist
        from meshbay_node.config import DEFAULT_STUN_SERVERS
        self._stun = stun_servers or list(DEFAULT_STUN_SERVERS)
        self._sessions: dict[str, WebRTCPeerSession] = {}

    async def handle_offer(
        self, offer_sdp: str, peer_id: str,
    ) -> tuple[str, list[dict]]:
        """
        Process a WebRTC SDP offer from a browser client.

        Returns (answer_sdp, ice_candidates) to relay back via hub signaling.
        ICE candidates are embedded in the SDP (aiortc gathers before returning).
        """
        from aiortc import RTCIceServer, RTCConfiguration

        # aiortc keeps only the first STUN entry it sees here; the actual
        # multi-server fan-out is done by transport/stun_multi, which patches
        # aioice. The full list is still passed so a one-server deploy and the
        # tests that read `_stun` stay coherent.
        config = RTCConfiguration(
            iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else []
        )
        pc = RTCPeerConnection(configuration=config)
        session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id)
        self._sessions[peer_id] = session

        @pc.on("datachannel")
        def on_datachannel(channel: RTCDataChannel):
            log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id)
            session._setup_channel(channel)

        if _WEBRTC_TRACE:
            @pc.on("iceconnectionstatechange")
            def on_ice_state_change():
                log.info("WebRTC ICE state: %s (peer=%s)", pc.iceConnectionState, peer_id)

        @pc.on("connectionstatechange")
        async def on_state_change():
            state = pc.connectionState
            log.info("WebRTC connection state: %s (peer=%s)", state, peer_id)
            if state in ("failed", "closed"):
                gone = self._sessions.pop(peer_id, None)
                if gone is not None:
                    # Popping only forgets the session. Its stream went on
                    # transcoding until the credit timeout — measured at 91s
                    # after the connection closed — holding one of the node's
                    # two slots the whole time. Closing the viewer, the tab or
                    # the browser all arrive here, so this is the one place
                    # that covers every way of walking away.
                    await gone.shutdown_tasks()

        offer = RTCSessionDescription(sdp=offer_sdp, type="offer")
        await pc.setRemoteDescription(offer)
        answer = await pc.createAnswer()
        gather_start = time.monotonic()
        await pc.setLocalDescription(answer)

        # ICE gathering runs inside setLocalDescription (non-trickle). A slow or
        # unreachable STUN server shows up here as seconds of wait and zero
        # srflx lines — the symptom the multi-server fan-out exists to prevent.
        answer_sdp = pc.localDescription.sdp
        srflx = answer_sdp.count(" typ srflx")
        # The host addresses this node put in the answer. When a peer reports
        # "DataChannel closed" the first question is whether the node offered
        # anything that peer could route to at all — on a NAT'd host or a VM the
        # only host candidate is an address no one else can reach, and the log
        # otherwise looks identical to a working connection.
        host_addrs: set[str] = set()
        for line in answer_sdp.splitlines():
            if line.startswith("a=candidate:") and " typ host " in line:
                parts = line.split()
                if len(parts) > 5:
                    host_addrs.add(parts[4])
        log.info(
            "WebRTC answer ready for peer=%s (ICE gather %.2fs, host: %s, %d srflx)",
            peer_id, time.monotonic() - gather_start,
            ", ".join(sorted(host_addrs)) or "none", srflx)
        return answer_sdp, []

    async def close_peer(self, peer_id: str) -> None:
        session = self._sessions.pop(peer_id, None)
        if session:
            await session.close()

    async def close_all(self) -> None:
        for session in list(self._sessions.values()):
            await session.close()
        self._sessions.clear()

    @property
    def active_peers(self) -> int:
        return len(self._sessions)