"""
MeshBay Node daemon — main process.

Startup sequence:
  1. Load config (~/.config/meshbay/node.toml)
  2. Load or create keystore (Argon2id unlock)
  3. Connect to hub: register → login → announce node
  4. Fetch GEK bundle from hub (if group configured)
  5. Start directory indexer (watchdog)
  6. Create chat stores (one SQLite DB per group)
  7. Create WebRTC transport (browser + native clients via DataChannel)
  8. Start QUIC chunk server (LAN / port-forwarded / hub-less direct access)
  9. (Phase 11.5: the unauthenticated HTTP file API and the TCP+TLS server were removed)
  10. Start hub WebSocket (signaling, revocations, WebRTC offers)
  11. Start local web UI on node.ui_port (localhost only)
  12. Run until SIGINT/SIGTERM

Usage:
  meshbay-node                    # interactive password prompt
  meshbay-node --config /path     # custom config
  meshbay-node status             # node state + public key (works while stopped)
  meshbay-node ui                 # print the local admin UI URL
  meshbay-node gek-init           # initialise the group key (no browser needed)
  meshbay-node init               # write example config + create keystore
  meshbay-node --calibrate-argon2 # benchmark Argon2id, suggest parameters
"""

import asyncio
from dataclasses import asdict, replace
import base64
import json
import logging
import os
import signal
import sys
from pathlib import Path

import uvicorn

from meshbay_common import MNP_VERSION
from meshbay_common.protocol import MNP, index_entry_wire
from meshbay_node.audit import AuditStore
from meshbay_node.bundle_store import BundleStore
from meshbay_node.chat.store import ChatStore
from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
from meshbay_node.roots import RootSet, RootError, entry_abs_path
from meshbay_node.hub_client import HubClient, HubConfig
from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex
from meshbay_node.indexer.enrich import Enricher
from meshbay_node.indexer.enrich_audio import AudioEnricher
from meshbay_node.indexer.enrich_photo import PhotoEnricher
from meshbay_node.media_cache import MediaCache
from meshbay_node.tmdb import TmdbClient
from meshbay_node.musicbrainz import MusicBrainzClient
from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
from meshbay_node.roster import Roster
from meshbay_node.transport import (
    Denylist,
    QUIC_AVAILABLE,
    WEBRTC_AVAILABLE,
)

if QUIC_AVAILABLE:
    from meshbay_node.transport import QuicChunkServer
if WEBRTC_AVAILABLE:
    from meshbay_node.transport import WebRTCTransport

log = logging.getLogger(__name__)


def _under_video_root(path: str, video_root: str) -> bool:
    """Mirrors video-app.js's underVideoRoot: same folder, or a descendant."""
    path = path or ""
    return path == video_root or path.startswith(video_root + "/")


def _under_audio_root(path: str, audio_root: str) -> bool:
    """Mirrors music-app.js's underAudioRoot — same shape as _under_video_root."""
    path = path or ""
    return path == audio_root or path.startswith(audio_root + "/")


def _under_any_photo_root(path: str, photo_roots: list[str]) -> bool:
    """
    Mirrors photos-app.js's underAnyPhotoRoot. Unlike video/audio's single
    root, photo_roots is a list (docs/photos.md §2.1) — a match against any
    one of them is enough.
    """
    path = path or ""
    return any(path == r or path.startswith(r + "/") for r in photo_roots)


# ── Argon2id calibration ──────────────────────────────────────────────────────

def calibrate_argon2(target_ms: int = 500) -> None:
    """Benchmark Argon2id and suggest parameters targeting ~target_ms."""
    import time
    import os

    print(f"Calibrating Argon2id (target: {target_ms}ms) ...")
    salt = os.urandom(16)
    for mem in [65536, 131072, 262144, 524288]:
        from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
        t0 = time.perf_counter()
        Argon2id(salt=salt, length=32, iterations=3,
                 lanes=4, memory_cost=mem).derive(b"benchmark")
        elapsed_ms = (time.perf_counter() - t0) * 1000
        print(f"  memory_cost={mem:>7} ({mem//1024:>4}MB): {elapsed_ms:.0f}ms", end="")
        if abs(elapsed_ms - target_ms) < target_ms * 0.3:
            print("  ← recommended")
        else:
            print()
    print("Set memory_cost in meshbay_common/crypto.py: ARGON2_MEMORY_COST")


# ── Hub WS sender bridge ─────────────────────────────────────────────────────

class _WsSender:
    """Thin bridge so WebRTC context can call hub_ws.send() for chat_notify."""

    def __init__(self, hub_client: HubClient):
        self._hub = hub_client

    async def send(self, data: str) -> None:
        await self._hub.send_ws(data)


# ── Daemon ────────────────────────────────────────────────────────────────────

class NodeDaemon:
    def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH):
        self._config = config
        self._config_path = config_path
        self._state: dict = {
            "status":        "starting",
            "hub_url":       config.hub.url,
            "username":      config.hub.username,
            "groups":        [g.name for g in config.groups],
            "quic_port":     config.node.quic_port,
            "endpoint_hint": None,
            "indexes":       {},
            "indexers":      {},
        }
        self._quic_server = None
        self._webrtc = None
        # Persisted so a restart does not silently un-revoke everyone (H4)
        self._denylist = (
            Denylist(path=config.data_dir / "denylist.json") if Denylist else None)
        self._chat_stores: dict[str, ChatStore] = {}
        # One instance, shared by every group's DirectoryIndexer — see
        # indexer/cache.py's docstring for why this stopped being per-group.
        self._index_cache: IndexCache | None = None
        # Coalesces a burst of index changes (one per debounced watchdog
        # event) into a single broadcast — see _on_index_change. 0.5s is
        # short enough nobody notices the wait, long enough that dropping a
        # few hundred files into a watched folder produces one push instead
        # of one per file.
        self._broadcast_coalesce_secs = 0.5
        self._pending_broadcasts: dict[str, asyncio.TimerHandle] = {}
        # group_id -> (version, {id: entry}) as of the last thing actually
        # broadcast — the comparison point for the next delta.
        self._last_broadcast_snapshot: dict[str, tuple] = {}
        self._audit_store: AuditStore | None = None
        self._bundle_store: BundleStore | None = None
        self._media_cache: MediaCache | None = None
        self._enricher: Enricher | None = None
        self._tmdb_client: TmdbClient | None = None
        self._audio_enricher: AudioEnricher | None = None
        self._musicbrainz_client: MusicBrainzClient | None = None
        self._photo_enricher: PhotoEnricher | None = None
        # A file id attempted at most once per daemon run, success or
        # failure — a persistently unprobeable file (corrupt, still being
        # written) does not get re-queued on every coalesced broadcast. A
        # restart retries everything, matching the "disposable, rebuildable"
        # stance the rest of this cache takes (docs/mediacenter.md §1/§2).
        # Shared across the video and audio enrichment paths — content-
        # addressed ids never collide between the two. Keyed by
        # (group_id, entry.id), not entry.id alone: the id is a content
        # hash, so the same physical file shared into two different groups
        # (found live — overlapping test libraries across several demo
        # groups) produces the same id in both. A bare-id set marked the
        # second group's copy "already attempted" the moment the first
        # group's enrichment ran, even though nothing had ever populated
        # *that* group's own index — every file in the second group stayed
        # at duration 0 with no artist/album, permanently, since nothing
        # ever revisits an id already in this set.
        self._enriched_attempted: set[tuple[str, str]] = set()
        self._roster: Roster | None = None
        self._indexers: list[DirectoryIndexer] = []
        self._tasks: list[asyncio.Task] = []
        self._hub: HubClient | None = None
        self._reload_lock = asyncio.Lock()

    async def run(self) -> None:
        log.info("MeshBay Node starting up")

        # 1. Keystore
        keys = load_or_create_keystore(
            path=self._config.keystore.path,
            unlock_file=self._config.keystore.unlock_file,
        )
        log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16])

        # 2. Start admin UI early (so operator can copy node key before hub login)
        self._state["pk_node_ed25519"] = keys.pk_ed25519_b64
        self._state["config"] = self._config
        # Where it came from, so `group add` appends to the file this process
        # actually read rather than guessing at the default.
        self._state["config_path"] = str(self._config_path)
        # Per-run token for the local admin UI (11.5.3). Not a password: it keeps
        # other local processes and rebound browser pages out of an API that can
        # re-initialise group keys.
        ui_token = base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=")
        self._state["ui_token"] = ui_token
        # Persisted so `meshbay-node ui` can open the browser. Nobody should ever
        # have to copy a token out of a log or a terminal — that is not a workflow.
        self._config.data_dir.mkdir(parents=True, exist_ok=True)
        self._ui_token_file = self._config.data_dir / "ui-token"
        self._ui_token_file.write_text(ui_token)
        self._ui_token_file.chmod(0o600)
        from meshbay_node.ui import create_ui_app
        ui_app = create_ui_app(self._state)
        ui_cfg = uvicorn.Config(
            ui_app,
            host="127.0.0.1",
            port=self._config.node.ui_port,
            log_level="warning",
        )
        ui_server = uvicorn.Server(ui_cfg)
        self._tasks.append(asyncio.create_task(ui_server.serve()))
        log.info("Admin UI ready — open it with:  meshbay-node ui")

        # 3. Hub connection (Ed25519 auth — retries until node key is linked)
        hub_cfg = HubConfig(
            hub_url=self._config.hub.url,
            username=self._config.hub.username,
        )
        async with HubClient(hub_cfg, keys) as hub:
            self._hub = hub
            session = await self._login_with_retry(hub)
            self._state["endpoint_hint"] = session.node_id

            try:
                session.email = await hub.fetch_owner_email()
            except Exception as e:
                log.warning("Could not fetch owner email: %s", e)

            # 4. Bundle store (P2P GEK bundles)
            data_dir = self._config.data_dir
            data_dir.mkdir(parents=True, exist_ok=True)
            self._bundle_store = BundleStore(db_path=data_dir / "bundles.db")
            await self._bundle_store.open()
            log.info("Bundle store opened: %s", data_dir / "bundles.db")

            # 4a. Path->hash cache, node-wide — opened once, shared by every
            # group's DirectoryIndexer below (indexer/cache.py).
            self._index_cache = IndexCache(db_path=data_dir / "index_cache.db")
            await self._index_cache.open()
            self._state["index_cache"] = self._index_cache
            log.info("Index cache opened: %s", data_dir / "index_cache.db")

            # 4b. Roster — who this node recognises and which keys are theirs.
            # Node authority is established here, locally, and never learned from
            # the hub: a hub that could name the operator's key could install
            # itself as node administrator.
            self._roster = Roster(db_path=data_dir / "roster.db")
            await self._roster.open()
            await self._roster.purge_expired()

            # Apply any roster overrides to node config (panel-edited values
            # take precedence over node.toml defaults).
            from meshbay_node.config import DEFAULT_STUN_SERVERS
            nd = self._config.node
            defaults = {
                "invite_ttl_hours": nd.invite_ttl_hours,
                "pair_ttl_hours": nd.pair_ttl_hours,
                "device_request_ttl_minutes": nd.device_request_ttl_minutes,
                "max_concurrent_streams": nd.max_concurrent_streams,
                "transcode_incompatible_video": nd.transcode_incompatible_video,
                "stun_servers": nd.stun_servers if nd.stun_servers else list(DEFAULT_STUN_SERVERS),
                "ice_interfaces": nd.ice_interfaces,
            }
            effective = await self._roster.node_settings(defaults)
            for k, v in effective.items():
                setattr(nd, k, v)

            # X25519 key material for GEK unwrapping
            from cryptography.hazmat.primitives import serialization
            sk_x_raw = keys.sk_x25519.private_bytes(
                serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
                serialization.NoEncryption())
            pk_x_raw = base64.b64decode(keys.pk_x25519_b64)

            # 4. Build per-group contexts
            groups_ctx: dict[str, dict] = {}
            for group_cfg in self._config.groups:
                if not group_cfg.id or not group_cfg.roots:
                    log.warning("Group %r has no id or no shared directory — "
                                "skipping", group_cfg.name)
                    continue

                try:
                    roots = RootSet.build([asdict(r) for r in group_cfg.roots])
                except RootError as e:
                    # Configuration the operator has to fix; guessing would put
                    # a member's file on the wrong disk or index one twice.
                    log.error("Group %r: %s — skipping", group_cfg.name, e)
                    continue

                roots.refresh_availability()
                if not any(r.available for r in roots):
                    # Not skipped for being empty: a group whose only drive is
                    # unplugged still exists, and its index is frozen rather
                    # than lost. But there is nothing to serve until it returns.
                    log.warning(
                        "Group %r: none of its %d root(s) are readable right now "
                        "(%s) — serving nothing until one returns",
                        group_cfg.name, len(roots),
                        ", ".join(str(r.path) for r in roots))

                gek = None
                gek = await self._load_gek(
                    group_cfg.id, session.user_id, sk_x_raw, pk_x_raw)
                if gek:
                    log.info("GEK loaded for group %s", group_cfg.id[:8])
                else:
                    log.info("No GEK yet for group %s — will accept first setup",
                             group_cfg.name)

                # Read once at load, like member_upload/enabled_apps below —
                # kept current in place afterwards by set_scan_settings
                # (ops.py), which updates both this indexer object directly
                # and roster.db, so a restart picks up the same values.
                scan_settings = (
                    await self._roster.scan_settings(group_cfg.id)
                    if self._roster else {
                        "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
                        "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
                    })

                indexer = DirectoryIndexer(
                    roots=roots,
                    group_id=group_cfg.id,
                    sk_node=keys.sk_ed25519,
                    gek=gek,
                    on_change=self._on_index_change,
                    cache=self._index_cache,
                    reconcile_secs=scan_settings["reconcile_interval_secs"],
                    debounce_secs=scan_settings["debounce_secs"],
                )
                await indexer.start(defer_scan=True)
                self._indexers.append(indexer)
                self._state["indexes"][group_cfg.id] = indexer.index
                self._state["indexers"][group_cfg.id] = indexer
                log.info("Group %s configured: %s (scan deferred)",
                         group_cfg.name,
                         ", ".join(f"{r.name}={r.path}" for r in roots))

                groups_ctx[group_cfg.id] = {
                    "gek": gek,
                    "roots": roots,
                    "index": indexer.index,
                    # Live reference, mutated in place by the indexer itself
                    # (see IndexProgress in indexer.py) — read, never copied,
                    # by the handshake ack and the periodic progress pusher.
                    "progress": indexer.progress,
                    # Bound method, called when a peer completes the
                    # handshake — resets reconcile's backoff (indexer.py
                    # _reconcile_loop) so the backstop is prompt again now
                    # that someone is actually looking.
                    "note_activity": indexer.note_activity,
                    # Shown to the operator in Settings, and kept current in
                    # place by set_scan_settings (ops.py) — same reasoning as
                    # member_upload below.
                    "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
                    "debounce_secs": scan_settings["debounce_secs"],
                    "visibility": group_cfg.visibility,
                    # Admission policy comes from node.toml, never from the hub:
                    # a hub that could declare a group open would be handed its key.
                    "join_policy": group_cfg.join_policy,
                    # Whether ordinary members may upload. Read once here, into
                    # the context, because the upload handler is synchronous and
                    # a database round trip per chunk would be absurd. The
                    # signed operation that changes it updates this dict in
                    # place, so the two never drift within a run.
                    "member_upload": await self._roster.member_upload_allowed(
                        group_cfg.id) if self._roster else True,
                    # Same reasoning: read once at load, kept current in place
                    # by the signed operation that changes it.
                    "enabled_apps": await self._roster.enabled_apps(
                        group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS),
                    # Which folder is the Videos app's entry point for this
                    # group — "" means the whole group index.
                    "video_root": await self._roster.video_root(
                        group_cfg.id) if self._roster else "",
                    # Same shape, Music app's own entry point.
                    "audio_root": await self._roster.audio_root(
                        group_cfg.id) if self._roster else "",
                    # Photos app's entry points — a *list*, unlike video_root/
                    # audio_root above (docs/photos.md §2.1: a photo library
                    # is routinely scattered across several folders). Empty
                    # list means nothing configured yet.
                    "photo_roots": await self._roster.photo_roots(
                        group_cfg.id) if self._roster else [],
                    # Whether TMDB lookups run for this group at all —
                    # 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.
                    "tmdb_enabled": await self._roster.tmdb_enabled(
                        group_cfg.id) if self._roster else True,
                    # Music app equivalent of tmdb_enabled — per-group from
                    # the start (docs/musicbay.md §6).
                    "musicbrainz_enabled": await self._roster.musicbrainz_enabled(
                        group_cfg.id) if self._roster else True,
                }

            if not groups_ctx:
                log.warning("No groups configured yet — admin UI and hub "
                            "connection stay up; attach a group to go live")

            # 5. Chat stores (one SQLite DB per group)
            for gid in groups_ctx:
                chat_db = data_dir / gid[:16] / "chat.db"
                store = ChatStore(db_path=chat_db)
                await store.open()
                self._chat_stores[gid] = store
                groups_ctx[gid]["chat_store"] = store
            log.info("Chat stores opened: %d groups", len(self._chat_stores))

            # 6. Audit store (legal compliance — IP + action logging)
            audit_db = data_dir / "audit.db"
            self._audit_store = AuditStore(db_path=audit_db)
            await self._audit_store.open()
            log.info("Audit store opened: %s", audit_db)

            # 6b. Media cache (Videos app — TMDB metadata + thumbnails).
            # Node-wide like audit.db, not per-group: a thumbnail is the same
            # bytes regardless of which group happens to share the file
            # (docs/mediacenter.md §2/§5.5).
            media_cache_db = data_dir / "media_cache.db"
            self._media_cache = MediaCache(db_path=media_cache_db)
            await self._media_cache.open()
            self._enricher = Enricher(self._media_cache)
            self._tmdb_client = TmdbClient(roster=self._roster)
            # Token/language only — read once at load, kept current in place
            # by ops.set_tmdb_config (the signed op), exposed to every
            # group's handshake ack via `daemon_state` (already wired to
            # self._webrtc._ctx below) since these stay node-wide, one
            # shared credential/cache. Whether TMDB is used at all is now
            # per-group instead — see each group's own "tmdb_enabled" in
            # groups_ctx above.
            tmdb_token, tmdb_language = await self._roster.tmdb_config()
            self._state["tmdb_token_customized"] = bool(tmdb_token)
            self._state["tmdb_language"] = tmdb_language or ""

            # 6c. Music app (docs/musicbay.md) — same media_cache.db, its own
            # enricher (mutagen, not ffmpeg) and its own MusicBrainz client.
            # The User-Agent contact is the owner's hub email, resolved at
            # login — no roster setting or env var needed any more.
            self._audio_enricher = AudioEnricher(self._media_cache)
            self._musicbrainz_client = MusicBrainzClient(owner_email=session.email)
            self._state["musicbrainz_contact_configured"] = bool(session.email)

            # 6d. Photos app (docs/photos.md) — same media_cache.db, its own
            # enricher (Pillow, not ffmpeg/mutagen). No credential, no
            # third-party client to construct: EXIF is read locally.
            self._photo_enricher = PhotoEnricher(self._media_cache)
            self._state["media_cache"] = self._media_cache
            log.info("Media cache opened: %s", media_cache_db)

            # 5. Denylist
            denylist = self._denylist

            # 6. WebRTC transport (browser clients)
            from meshbay_node.transport.ice_filter import install as install_ice_filter
            install_ice_filter(
                self._config.node.ice_interfaces or None,
            )
            # aiortc keeps only the FIRST entry of RTCConfiguration.iceServers, so
            # the multi-STUN fallback only exists on the node if aioice itself
            # fans out — see transport/stun_multi.
            from meshbay_node.transport.stun_multi import install as install_stun_multi
            install_stun_multi(self._config.node.stun_servers or None)

            first = next(iter(groups_ctx.values()), None)
            if WEBRTC_AVAILABLE:
                self._webrtc = WebRTCTransport(
                    sk_node=keys.sk_ed25519,
                    hub_pk_pem=session.hub_pk_pem,
                    gek=first["gek"] if first else None,
                    roots=first["roots"] if first else None,
                    index=first["index"] if first else None,
                    groups=groups_ctx,
                    denylist=denylist,
                    max_concurrent_streams=self._config.node.max_concurrent_streams,
                    transcode_incompatible_video=self._config.node.transcode_incompatible_video,
                    stun_servers=self._config.node.stun_servers or None,
                )
                # No global chat_store here: each group's store lives in
                # groups_ctx[gid]["chat_store"] and is resolved per session via
                # _group_ctx(). Assigning the first group's store transport-wide
                # sent every group's chat to one database and served it back to
                # members of every other group (finding H1).
                self._webrtc._ctx["groups"] = groups_ctx
                self._webrtc._ctx["hub_ws"] = _WsSender(hub)
                self._webrtc._ctx["node_user_id"] = session.user_id
                self._webrtc._ctx["audit_store"] = self._audit_store
                self._webrtc._ctx["bundle_store"] = self._bundle_store
                self._webrtc._ctx["media_cache"] = self._media_cache
                self._webrtc._ctx["tmdb_client"] = self._tmdb_client
                self._webrtc._ctx["musicbrainz_client"] = self._musicbrainz_client
                self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw
                self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw
                self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64

                self._webrtc._ctx["roster"] = self._roster
                # The MNP adapter calls the same operations as the loopback API
                # (meshbay_node.ops), and those take the daemon's state. Handing
                # the transport a second set of lookups is how two paths to one
                # operation start disagreeing — the shape of C1 and C6.
                self._webrtc._ctx["daemon_state"] = self._state
                self._webrtc._ctx["invite_ttl"] = (
                    self._config.node.invite_ttl_hours * 3600)
                self._webrtc._ctx["device_request_ttl"] = (
                    self._config.node.device_request_ttl_minutes * 60)
                paired = await self._roster.has_operator() if self._roster else False
                self._webrtc._ctx["has_admin_authority"] = paired
                if paired:
                    log.info("Node authority: paired operator")
                else:
                    log.warning(
                        "No operator paired — invites and file deletion are "
                        "refused. Run: meshbay-node operator pair")
                log.info("WebRTC transport ready")
            else:
                log.warning("WebRTC not available (aiortc not installed)")

            # 7. QUIC chunk server (LAN / port-forwarded / hub-less direct access)
            if QUIC_AVAILABLE:
                self._quic_server = QuicChunkServer(
                    sk_node=keys.sk_ed25519,
                    hub_pk_pem=session.hub_pk_pem,
                    gek=first["gek"] if first else None,
                    roots=first["roots"] if first else None,
                    index=first["index"] if first else None,
                    host="::",
                    port=self._config.node.quic_port,
                    groups=groups_ctx,
                    denylist=denylist,
                )
                self._quic_server._ctx["groups"] = groups_ctx
                await self._quic_server.start()
                log.info("QUIC server on port %d (%d groups)",
                         self._config.node.quic_port, len(groups_ctx))

            # 8. Hub WebSocket (signaling + revocations + WebRTC offers)
            async def on_webrtc_offer(sdp, peer_id, ice_candidates):
                if not self._webrtc:
                    return None
                try:
                    answer_sdp, answer_ice = await self._webrtc.handle_offer(
                        sdp, peer_id)
                    log.info("WebRTC answer for peer=%s (%d peers)",
                             peer_id, self._webrtc.active_peers)
                    return (answer_sdp, answer_ice)
                except Exception as e:
                    log.error("WebRTC offer failed: %s", e)
                    return None

            async def on_incoming(peer_ip, peer_port):
                if self._quic_server:
                    self._quic_server.punch_nat(peer_ip, peer_port)

            def on_revocation(token):
                if denylist and token:
                    import jwt as _jwt
                    try:
                        payload = _jwt.decode(
                            token, session.hub_pk_pem, algorithms=["EdDSA"],
                            options={"verify_exp": False})
                        target = payload.get("target")
                        tid = payload.get("target_id", "")
                        if target == "user":
                            denylist.deny_user(tid)
                        elif target == "group":
                            # H4: previously dropped on the floor, so "suspend a
                            # group" was a hub-only gesture that no node enforced.
                            denylist.deny_group(tid)
                            self._drop_group_sessions(tid)
                        elif target == "jti":
                            denylist.deny_jti(tid)
                        else:
                            log.warning("Unknown revocation target: %r", target)
                    except Exception as e:
                        log.warning("Invalid revocation token: %s", e)

            ws_task = asyncio.create_task(hub.maintain_ws(
                on_incoming=on_incoming,
                on_revocation=on_revocation,
                on_webrtc_offer=on_webrtc_offer,
                group_ids=lambda: list((self._state.get("groups_ctx") or {}).keys()),
            ))
            self._tasks.append(ws_task)
            log.info("Hub WS task started")

            # 9. (removed in Phase 11.5) The per-group HTTP file API used to start here.
            # It served the Mesh Group Index and raw plaintext files on 0.0.0.0 with no
            # authentication, for private groups too — finding C1. Every client path now
            # goes through the MNP handshake (JWT + group claim + GEK proof).

            # 10. Update admin UI state (UI already running from step 2)
            self._state["groups_ctx"] = groups_ctx
            self._state["audit_store"] = self._audit_store
            self._state["bundle_store"] = self._bundle_store
            self._state["roster"] = self._roster
            self._state["node_user_id"] = session.user_id
            self._state["webrtc"] = self._webrtc
            self._state["quic_server"] = self._quic_server
            self._state["hub"] = hub
            self._state["reload_fn"] = self._reload_config
            self._state["enrich_video_root_fn"] = self._enrich_video_root_now
            self._state["enrich_audio_root_fn"] = self._enrich_audio_root_now
            self._state["enrich_photo_roots_fn"] = self._enrich_photo_roots_now
            # Rotating a key has to reach every transport holding a copy of it,
            # and clearing the denylist has to reach the one the handshake
            # consults — so both are published rather than reachable only
            # through the object that happens to own them.
            self._state["denylist"] = self._denylist
            self._state["pk_x25519_raw"] = pk_x_raw
            self._state["sk_x25519_raw"] = sk_x_raw
            self._state["sk_ed25519"] = keys.sk_ed25519

            self._state["status"] = "running"
            log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s",
                     len(groups_ctx),
                     "yes" if self._webrtc else "no",
                     "yes" if self._quic_server else "no")

            # 11. Background initial scan — files appear progressively.
            async def _bg_scan(indexer, name, gctx):
                await indexer.initial_scan()
                log.info("Background scan complete for %s: %d files",
                         name, indexer.index.count)
                # Swarm registration for public groups (after files are known).
                if gctx.get("visibility") == "public":
                    endpoint = f"webrtc:{self._config.node.quic_port}"
                    hashes = [e.id for e in gctx["index"].entries]
                    if hashes:
                        await self._register_swarm(hashes, endpoint)
                # initial_scan() itself never calls on_change (it predates
                # the concept — every existing caller only cared about the
                # scan finishing, not about notifying anyone) — but Videos
                # app enrichment (duration/thumb_hash/display_title/...)
                # hangs entirely off that callback (_broadcast_index_change).
                # Without this, every file already on disk at startup — the
                # common case, an existing library — would never get
                # enriched at all; only a file added later, while the node
                # is already running, would trigger it via the watchdog.
                await self._on_index_change(indexer)

            for idx, group_cfg in zip(self._indexers, self._config.groups):
                gctx = groups_ctx.get(group_cfg.id)
                if gctx:
                    self._tasks.append(asyncio.create_task(
                        _bg_scan(idx, group_cfg.name, gctx)))
                    self._tasks.append(asyncio.create_task(
                        self._progress_pusher(idx)))

            # 12. Wait for shutdown
            stop_event = asyncio.Event()
            loop = asyncio.get_event_loop()
            for sig in (signal.SIGINT, signal.SIGTERM):
                loop.add_signal_handler(sig, stop_event.set)
            # Milestone 14.8: re-read node.toml without dropping connections.
            try:
                loop.add_signal_handler(
                    signal.SIGHUP,
                    lambda: asyncio.ensure_future(self._reload_config()))
            except (NotImplementedError, AttributeError):
                pass          # no SIGHUP on Windows; `reload` says so there
            await stop_event.wait()

        await self._shutdown()

    async def _reload_config(self) -> None:
        """
        Re-read node.toml and reconcile groups.

        Handles root changes on existing groups, hot-loads new groups, and
        tears down removed groups. Existing connections are untouched: a member
        watching a film keeps watching it.

        Serialised by _reload_lock: fire-and-forget reloads from config-mutating
        endpoints can overlap with the wizard's explicit /api/reload call,
        and two concurrent hot-loads of the same group corrupt the runtime state.
        """
        async with self._reload_lock:
            await self._reload_config_inner()

    async def _reload_config_inner(self) -> None:
        log.info("Reloading config from %s", self._config_path)
        try:
            fresh = load_config(self._config_path)
        except Exception as e:
            log.error("Reload failed, keeping the running config: %s", e)
            return

        groups_ctx = self._state.get("groups_ctx", {})
        hosted = set(groups_ctx)
        incoming = {g.id for g in fresh.groups if g.id}

        # ── Root changes on existing groups ──────────────────────────────
        changed = 0
        for group_cfg in fresh.groups:
            ctx = groups_ctx.get(group_cfg.id)
            if not ctx:
                continue
            try:
                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
            except RootError as e:
                log.error("Group %r: %s — keeping the roots already loaded",
                          group_cfg.name, e)
                continue
            before = {(r.name, str(r.path)) for r in ctx["roots"]}
            after = {(r.name, str(r.path)) for r in roots}
            if before == after:
                continue
            roots.refresh_availability()
            indexer = next((i for i in self._indexers
                            if i.group_id == group_cfg.id), None)
            if indexer is None:
                continue
            log.info("Group %r roots changed: %s", group_cfg.name,
                     ", ".join(f"{r.name}={r.path}" for r in roots))
            await indexer.retarget(roots)
            ctx["roots"] = roots
            changed += 1

        # ── Hot-load new groups ──────────────────────────────────────────
        added_names = []
        sk_ed = self._state.get("sk_ed25519")
        sk_x_raw = self._state.get("sk_x25519_raw")
        pk_x_raw = self._state.get("pk_x25519_raw")
        node_user_id = self._state.get("node_user_id")
        data_dir = fresh.data_dir

        for group_cfg in fresh.groups:
            if group_cfg.id in hosted:
                continue
            if not group_cfg.id or not group_cfg.roots:
                log.warning("New group %r has no id or roots — skipping",
                            group_cfg.name)
                continue
            if not sk_ed:
                log.warning("Cannot hot-load %r — signing key not available",
                            group_cfg.name)
                continue

            try:
                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
            except RootError as e:
                log.error("New group %r: %s — skipping", group_cfg.name, e)
                continue
            roots.refresh_availability()

            gek = None
            if sk_x_raw and pk_x_raw:
                gek = await self._load_gek(
                    group_cfg.id, node_user_id, sk_x_raw, pk_x_raw)
                if gek:
                    log.info("GEK loaded for new group %s", group_cfg.id[:8])

            scan_settings = (
                await self._roster.scan_settings(group_cfg.id)
                if self._roster else {
                    "reconcile_interval_secs": DirectoryIndexer.DEFAULT_RECONCILE_SECS,
                    "debounce_secs": DirectoryIndexer.DEFAULT_DEBOUNCE_SECS,
                })

            indexer = DirectoryIndexer(
                roots=roots,
                group_id=group_cfg.id,
                sk_node=sk_ed,
                gek=gek,
                on_change=self._on_index_change,
                cache=self._index_cache,
                reconcile_secs=scan_settings["reconcile_interval_secs"],
                debounce_secs=scan_settings["debounce_secs"],
            )
            # Registered *before* start() runs its (blocking, possibly very
            # long — see the StarWars benchmark) initial scan, specifically
            # so /api/groups/{id}/index-status can see indexer.progress
            # while a brand-new group is still scanning — this is the one
            # group state that must stay visible during the very window the
            # group is not yet authorized for member connections (below).
            self._indexers.append(indexer)
            self._state["indexes"][group_cfg.id] = indexer.index
            self._state["indexers"][group_cfg.id] = indexer

            await indexer.start()

            data_dir.mkdir(parents=True, exist_ok=True)
            chat_db = data_dir / group_cfg.id[:16] / "chat.db"
            store = ChatStore(db_path=chat_db)
            await store.open()
            self._chat_stores[group_cfg.id] = store

            new_ctx = {
                "gek": gek,
                "roots": roots,
                "index": indexer.index,
                "progress": indexer.progress,
                "note_activity": indexer.note_activity,
                "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
                "debounce_secs": scan_settings["debounce_secs"],
                "visibility": group_cfg.visibility,
                "join_policy": group_cfg.join_policy,
                "member_upload": (
                    await self._roster.member_upload_allowed(group_cfg.id)
                    if self._roster else True),
                "enabled_apps": (
                    await self._roster.enabled_apps(group_cfg.id)
                    if self._roster else list(Roster.DEFAULT_APPS)),
                "video_root": (
                    await self._roster.video_root(group_cfg.id)
                    if self._roster else ""),
                "audio_root": (
                    await self._roster.audio_root(group_cfg.id)
                    if self._roster else ""),
                "photo_roots": (
                    await self._roster.photo_roots(group_cfg.id)
                    if self._roster else []),
                "tmdb_enabled": (
                    await self._roster.tmdb_enabled(group_cfg.id)
                    if self._roster else True),
                "musicbrainz_enabled": (
                    await self._roster.musicbrainz_enabled(group_cfg.id)
                    if self._roster else True),
                "chat_store": store,
            }
            groups_ctx[group_cfg.id] = new_ctx

            if self._webrtc:
                self._webrtc._ctx["groups"][group_cfg.id] = new_ctx
            log.info("Hot-loaded group %s (%s, %d roots)",
                     group_cfg.name, group_cfg.id[:8], len(roots))
            added_names.append(group_cfg.name)
            # The initial scan above already ran to completion (indexer.start()
            # is not deferred here), so this only matters for whatever scans
            # this group as time goes on — a root added later, reconcile
            # picking one back up.
            self._tasks.append(asyncio.create_task(self._progress_pusher(indexer)))

        # ── Tear down removed groups ─────────────────────────────────────
        removed_names = []
        for gid in hosted - incoming:
            indexer = next((i for i in self._indexers
                            if i.group_id == gid), None)
            if indexer:
                try:
                    await indexer.stop()
                except Exception:
                    pass
                self._indexers.remove(indexer)
            store = self._chat_stores.pop(gid, None)
            if store:
                try:
                    await store.close()
                except Exception:
                    pass
            # No per-group index cache to close here (2026-08-25): the
            # (path, size, mtime) -> hash cache is now one shared instance,
            # open for the life of the daemon, since another group may still
            # reference the same physical folder — see indexer/cache.py.
            pending = self._pending_broadcasts.pop(gid, None)
            if pending:
                pending.cancel()
            self._last_broadcast_snapshot.pop(gid, None)
            self._state["indexes"].pop(gid, None)
            self._state["indexers"].pop(gid, None)
            old_name = gid[:8]
            for g_cfg in self._config.groups:
                if g_cfg.id == gid:
                    old_name = g_cfg.name
                    break
            groups_ctx.pop(gid, None)
            if self._webrtc and self._webrtc._ctx.get("groups") is not groups_ctx:
                self._webrtc._ctx["groups"].pop(gid, None)
            log.info("Unloaded group %s (%s)", old_name, gid[:8])
            removed_names.append(old_name)

        self._config = fresh
        self._state["config"] = fresh
        self._state["groups"] = [g.name for g in fresh.groups]
        log.info("Reload complete — %d re-rooted, %d added, %d removed",
                 changed, len(added_names), len(removed_names))

        if (added_names or removed_names) and self._hub:
            gids = list((self._state.get("groups_ctx") or {}).keys())
            await self._hub.update_ws_groups(gids)

    async def _login_with_retry(self, hub: HubClient):
        """Login to hub, retrying if the node key hasn't been linked yet."""
        import httpx as _httpx
        while True:
            try:
                return await hub.startup(endpoint_hint=None)
            except _httpx.HTTPStatusError as e:
                body = e.response.text if hasattr(e.response, 'text') else ''
                # Any 401 here needs a human at a browser, and the operator needs
                # this daemon alive to read its public key out of the local admin
                # UI. Exiting would take that UI down and strand them — which is
                # exactly what happened when a node was started before its owner
                # had registered.
                if e.response.status_code == 401:
                    if "No node key" in body:
                        self._state["status"] = "waiting_for_node_key"
                        log.warning(
                            "Node key not linked. Open the admin UI, copy this "
                            "node's key, and paste it in Settings > Link Node on "
                            "%s. Retrying in 5s...",
                            self._config.hub.url,
                        )
                    else:
                        self._state["status"] = "waiting_for_account"
                        log.warning(
                            "Hub rejected the node credentials for user %r. "
                            "Register that account on %s first, then link this "
                            "node's key. Retrying in 5s...",
                            self._config.hub.username, self._config.hub.url,
                        )
                    await asyncio.sleep(5)
                else:
                    raise
            except Exception as e:
                log.warning("Hub login failed: %s — retrying in 10s", e)
                await asyncio.sleep(10)

    async def _load_gek(
        self,
        group_id: str,
        node_user_id: str,
        sk_x_raw: bytes,
        pk_x_raw: bytes,
    ) -> bytes | None:
        """Load GEK from local bundle store (node-only, hub never touches crypto)."""
        from meshbay_common.crypto import unwrap_gek_aes

        if not self._bundle_store:
            return None

        # Try node-specific bundle first (stored by init_gek for daemon reload),
        # then fall back to operator's user bundle (legacy / pre-dual-key)
        for user_key in [f"_node_{node_user_id}", node_user_id]:
            bundle = await self._bundle_store.fetch(group_id, user_key)
            if not bundle:
                continue
            try:
                gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw)
                log.info("GEK loaded from local bundle store for group %s (key=%s)",
                         group_id[:8], user_key[:16])
                return gek
            except Exception as e:
                log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e)

        log.warning("No unwrappable GEK bundle found for group %s", group_id[:8])
        return None

    async def _progress_pusher(self, indexer: DirectoryIndexer,
                               interval: float = 2.0) -> None:
        """
        Watches indexer.progress and pushes a light INDEX_PROGRESS message to
        this group's connected peers — never the index itself, that stays
        _on_index_change's job. Runs for the node's whole lifetime: a scan
        can start from several places (initial scan, a root added later,
        reconcile picking a root back up), and this only needs to notice the
        flag, not why it changed.

        The final push at the False transition is what lets a presence dot
        reliably turn back off on an already-connected client — the
        handshake ack only covers the moment of connecting. `interval` is a
        parameter (not a bare constant) only so a test can drive this loop
        without waiting on the real 2s cadence.
        """
        was_scanning = False
        while True:
            await asyncio.sleep(interval)
            progress = indexer.progress
            now_scanning = progress.scanning
            if now_scanning or was_scanning:
                self._push_index_progress(indexer.group_id, progress)
            was_scanning = now_scanning

    def _push_index_progress(self, group_id: str, progress) -> None:
        if not self._webrtc:
            return
        msg = {
            "type": MNP.INDEX_PROGRESS,
            "v": MNP_VERSION,
            "group_id": group_id,
            "scanning": progress.scanning,
            "scanned_bytes": progress.scanned_bytes,
            "total_bytes": progress.total_bytes,
        }
        pushed = 0
        for session in list(self._webrtc._sessions.values()):
            if session._group_id == group_id:
                try:
                    session._send(msg)
                    pushed += 1
                except Exception:
                    pass
        if pushed:
            log.debug("Index progress pushed to %d peer(s) for group %s",
                      pushed, group_id[:8])

    async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
        """
        Called when a DirectoryIndexer detects file changes — once per
        debounced watchdog event, so dropping N files into a watched folder
        calls this N times in quick succession. Coalesces those into one
        broadcast (_broadcast_index_change) rather than one push per file:
        the timer is reset on every call and only fires once calls stop
        arriving for _broadcast_coalesce_secs.
        """
        group_id = indexer.group_id
        loop = asyncio.get_event_loop()
        pending = self._pending_broadcasts.pop(group_id, None)
        if pending:
            pending.cancel()

        def fire() -> None:
            self._pending_broadcasts.pop(group_id, None)
            asyncio.ensure_future(self._broadcast_index_change(indexer))

        self._pending_broadcasts[group_id] = loop.call_later(
            self._broadcast_coalesce_secs, fire)

    async def _broadcast_index_change(self, indexer: DirectoryIndexer) -> None:
        """
        The actual push, run once per coalesced burst. Sends a full
        INDEX_SYNC the first time a group is ever broadcast (no previous
        snapshot to diff against — the client's own first fetchIndex() call
        already covers that case) and an INDEX_DELTA every time after,
        computed against the last thing this method actually sent.
        """
        group_id = indexer.group_id
        idx = indexer.index
        log.info("Index changed for group %s: %d files (v%d)",
                 group_id[:8], idx.count, idx.version)

        prev = self._last_broadcast_snapshot.get(group_id)
        delta = None
        previous = None
        if prev is not None:
            prev_version, prev_entries = prev
            previous = GroupIndex._snapshot(
                idx.group_id, idx.sk_node, idx.gek, prev_version, prev_entries)
            delta = idx.diff(previous)
        self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id())

        # Videos app (docs/mediacenter.md §5.2): schedule async technical
        # probe + title parse + thumbnail generation for every newly-seen
        # video entry under the group's configured video_root. Never blocks
        # this broadcast — enrichment fields arrive later as their own
        # INDEX_DELTA update (_on_enriched below).
        new_entries = delta.additions if delta is not None else list(idx.entries)
        asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries))
        # Music app (docs/musicbay.md §6): same shape, gated on audio_root
        # exactly like video_root above (added later — musicbay.md's
        # original "no root, whole shared tree" call didn't hold up).
        asyncio.ensure_future(self._enrich_new_audio_entries(indexer, new_entries))
        # Photos app (docs/photos.md §5): same shape, gated on photo_roots
        # (a list, not a single string — §2.1).
        asyncio.ensure_future(self._enrich_new_photo_entries(indexer, new_entries))

        # A rename/move changes the very filename (or season folder) that
        # §3.3/§3.4's title-parse read display_title/season/episode from,
        # but leaves the file's content — and so its id and everything
        # ffprobe/thumbnailing already found — untouched. Only entries
        # whose name or path actually differ from the last broadcast get a
        # fresh pass; an update that is enrichment's own field-fill
        # (duration/thumb_hash/... landing via _on_enriched below) leaves
        # name/path alone and must not re-trigger itself forever.
        if delta is not None and delta.updates and previous is not None:
            asyncio.ensure_future(
                self._reenrich_renamed_video_entries(indexer, delta.updates, previous))
            asyncio.ensure_future(
                self._reenrich_renamed_audio_entries(indexer, delta.updates, previous))
            asyncio.ensure_future(
                self._reenrich_renamed_photo_entries(indexer, delta.updates, previous))

        # Videos/Music/Photos apps: a file that leaves the index also loses
        # its thumbnail/cover and file->tmdb/file->mbid mapping — the "real
        # deletion obligation" docs/mediacenter.md §2/§8 calls out
        # explicitly rather than leaving implicit (docs/musicbay.md §6
        # follows the same rule). tmdb_meta/mbid_meta rows are left alone
        # (§2: shared across files).
        #
        # Found live (docs/photos.md): a root removed and a new one added
        # for the identical content (an operator renaming/relocating a
        # shared folder) pruned the thumbnail here — correctly, the content
        # is gone from *this* root — but left the hash in
        # `_enriched_attempted`, which is never otherwise cleared. The same
        # bytes reappearing under the new root's path were then permanently
        # skipped: "already attempted" was true forever, for a thumbnail
        # that no longer existed. Discarding the attempt alongside the
        # cache entry is what makes pruning actually reversible — the next
        # sweep re-enriches it exactly as if it were new, which content
        # that is content-addressed and simply moved effectively is.
        if delta is not None and delta.deletions and self._media_cache:
            for file_id in delta.deletions:
                self._enriched_attempted.discard((indexer.group_id, file_id))
                # media_cache.db is node-wide, keyed by content hash — a file
                # shared into two groups is one row there, same reasoning as
                # `_enriched_attempted`'s own docstring above. This group's
                # copy is genuinely gone (that's what a deletion delta is),
                # but another group may still hold the same content: only
                # prune once *no* group's index has this file_id any more,
                # or the surviving group pays for a redundant re-fetch/
                # re-probe/re-thumbnail for content it never actually lost.
                still_referenced = any(
                    i.index.get_entry(file_id) is not None for i in self._indexers)
                if not still_referenced:
                    asyncio.ensure_future(self._media_cache.prune_file(file_id))

        # 11.5 — Push to connected WebRTC peers in this group
        if self._webrtc:
            if delta is not None:
                msg = {
                    "type": MNP.INDEX_DELTA,
                    "v": MNP_VERSION,
                    "group_id": idx.group_id,
                    "base_version": delta.base_version,
                    "version": delta.version,
                    "additions": [index_entry_wire(e) for e in delta.additions],
                    "deletions": delta.deletions,
                    "updates": [index_entry_wire(e) for e in delta.updates],
                }
            else:
                msg = {
                    "type": MNP.INDEX_SYNC,
                    "v": MNP_VERSION,
                    "group_id": idx.group_id,
                    "version": idx.version,
                    "entries": [index_entry_wire(e) for e in idx.entries],
                }
            pushed = 0
            for session in list(self._webrtc._sessions.values()):
                if session._group_id == group_id:
                    try:
                        session._send(msg)
                        pushed += 1
                    except Exception:
                        pass
            if pushed:
                log.info("Index %s pushed to %d WebRTC peers",
                         "delta" if delta is not None else "sync", pushed)

        # 11.9 — Register file hashes with hub swarm table (public groups only, H7)
        group_cfg = next(
            (g for g in self._config.groups if g.id == group_id), None)
        if (self._hub and self._state.get("endpoint_hint")
                and group_cfg and group_cfg.visibility == "public"):
            # Only the newly added hashes once there is a delta to know them
            # from — registering the whole library again on every change is
            # the same O(changes x library size) cost the delta above exists
            # to avoid.
            hashes = ([e.id for e in delta.additions] if delta is not None
                     else [e.id for e in idx.entries])
            if hashes:
                endpoint = f"webrtc:{self._config.node.quic_port}"
                asyncio.ensure_future(self._register_swarm(hashes, endpoint))

    async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
        """
        Videos app: fire (never await further) enrichment for unattempted
        video entries under the group's configured video_root.

        A group with no video_root set yet does not enrich anything — TMDB
        lookups and ffmpeg thumbnailing are real, ongoing per-file cost
        (mediacenter.md §5.2/§10), and running them over an operator's whole
        shared index before they have chosen which folder is actually their
        media library would burn both TMDB's rate limit and the node's CPU
        on files that were never meant to be in the Videos app at all. Once a
        root is set, `_enrich_video_root_now` (called from ops.set_video_root)
        separately sweeps whatever it already contains — this path alone only
        ever sees entries new since the last broadcast.
        """
        if not self._enricher or not self._roster:
            return
        video_root = await self._roster.video_root(indexer.group_id)
        if not video_root:
            return
        for entry in entries:
            if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted:
                continue
            if not _under_video_root(entry.path, video_root):
                continue
            file_path = entry_abs_path(indexer.roots, entry)
            if not file_path or not file_path.exists():
                continue
            self._enriched_attempted.add((indexer.group_id, entry.id))

            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
                await self._on_enriched(_indexer, file_id, fields)

            self._enricher.spawn(entry, file_path, on_done)

    async def _enrich_video_root_now(self, group_id: str) -> None:
        """
        Videos app: sweep a group's existing index for enrichment right
        after its video_root is set or changed (ops.set_video_root).

        The ordinary path above only ever looks at entries new since the
        last broadcast, so a folder that already had files sitting in it
        before it became the video_root would otherwise never get enriched
        at all — nothing else re-visits already-indexed entries once they
        have been broadcast once.
        """
        indexer = self._state.get("indexers", {}).get(group_id)
        if not indexer:
            return
        await self._enrich_new_video_entries(indexer, list(indexer.index.entries))

    async def _reenrich_renamed_video_entries(
        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
    ) -> None:
        """
        Videos app: found live — a French-named episode file, renamed by
        the operator to match its English-named siblings, kept showing as
        its own separate poster-grid card (and its own row in Flat list)
        indefinitely, because `_enriched_attempted` — there specifically to
        stop enrichment's own field-fill from re-triggering itself forever
        (see the caller) — also silently blocked the *new* filename from
        ever being title-parsed at all. `entry.id in self._enriched_attempted`
        is the same content, so simply discarding it here and re-running
        the ordinary enrichment path is enough: a fresh ffprobe/thumbnail
        for an unchanged file is redundant work, not a correctness issue,
        and renames are rare enough that the redundancy is not worth a
        separate "title-parse only" code path.
        """
        for entry in updates:
            if entry.type != "video":
                continue
            old = previous.get_entry(entry.id)
            if old is None or (old.name == entry.name and old.path == entry.path):
                continue
            self._enriched_attempted.discard((indexer.group_id, entry.id))
            # The rename re-derives the title (the whole point of this
            # method), which can change the correct TMDB match — but
            # file_tmdb is keyed by content hash, unchanged by a rename, so
            # nothing else would ever dislodge the old name's match. A
            # manual "Fix match" correction is kept (clear_file_tmdb skips
            # anything in media_cache.tmdb_override).
            if self._media_cache is not None:
                await self._media_cache.clear_file_tmdb(entry.id)
        await self._enrich_new_video_entries(indexer, updates)

    async def _enrich_new_audio_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
        """
        Music app (docs/musicbay.md §2.1, §6): fire (never await further)
        tag/cover enrichment for unattempted audio entries under the
        group's configured audio_root — same gate as
        `_enrich_new_video_entries` above (musicbay.md's original "no root,
        whole shared tree" call turned out wrong against a real messy
        library: everything under every shared folder got mixed together
        with no way to scope it down). `_enriched_attempted` is shared with
        the video path — content-addressed ids never collide across the two.
        """
        if not self._audio_enricher or not self._roster:
            return
        audio_root = await self._roster.audio_root(indexer.group_id)
        if not audio_root:
            return
        root_boundary = indexer.roots.resolve(audio_root, require_available=False)
        for entry in entries:
            if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted:
                continue
            if not _under_audio_root(entry.path, audio_root):
                continue
            file_path = entry_abs_path(indexer.roots, entry)
            if not file_path or not file_path.exists():
                continue
            self._enriched_attempted.add((indexer.group_id, entry.id))

            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
                await self._on_enriched(_indexer, file_id, fields)

            # `root_boundary` — audio_root itself, not the shared root it
            # lives under — so the ancestor walk
            # (enrich_audio._artist_album_from_ancestors) treats a flat
            # top-level folder right under the *configured* Music root as
            # ambiguous (artist-or-release, §2.1), not one level too shallow
            # if audio_root is itself a subfolder of a larger shared root.
            self._audio_enricher.spawn(entry, file_path, on_done, root_boundary)

    async def _enrich_audio_root_now(self, group_id: str) -> None:
        """
        Music app: sweep a group's existing index right after its
        audio_root is set or changed (ops.set_audio_root). Mirrors
        `_enrich_video_root_now` exactly — the ordinary path above only
        ever looks at entries new since the last broadcast, so a folder
        that already had files in it before it became the audio_root would
        otherwise never get enriched at all.
        """
        indexer = self._state.get("indexers", {}).get(group_id)
        if not indexer:
            return
        await self._enrich_new_audio_entries(indexer, list(indexer.index.entries))

    async def _reenrich_renamed_audio_entries(
        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
    ) -> None:
        """
        Music app equivalent of `_reenrich_renamed_video_entries` — a rename
        can change the filename-parse fallback (title/track_no) even though
        embedded tags, when present, are unaffected. Re-running the whole
        pass on a rename is redundant work for a tagged file and a real fix
        for an untagged one, and renames are rare enough not to need a
        cheaper, tags-only special case.
        """
        for entry in updates:
            if entry.type != "audio":
                continue
            old = previous.get_entry(entry.id)
            if old is None or (old.name == entry.name and old.path == entry.path):
                continue
            self._enriched_attempted.discard((indexer.group_id, entry.id))
        await self._enrich_new_audio_entries(indexer, updates)

    async def _enrich_new_photo_entries(self, indexer: DirectoryIndexer, entries: list) -> None:
        """
        Photos app (docs/photos.md §5): fire (never await further) thumbnail/
        EXIF enrichment for unattempted image entries under any of the
        group's configured photo_roots. Same gate shape as
        `_enrich_new_video_entries`/`_enrich_new_audio_entries` — no root
        configured yet means no work, since thumbnailing every image in a
        whole shared tree before the operator has chosen which folders are
        actually photo albums would burn CPU on files never meant to be in
        the Photos app at all. `_enriched_attempted` is shared with the
        video/audio paths — content-addressed ids never collide across them.
        """
        if not self._photo_enricher or not self._roster:
            return
        photo_roots = await self._roster.photo_roots(indexer.group_id)
        if not photo_roots:
            return
        for entry in entries:
            if entry.type != "image" or (indexer.group_id, entry.id) in self._enriched_attempted:
                continue
            if not _under_any_photo_root(entry.path, photo_roots):
                continue
            file_path = entry_abs_path(indexer.roots, entry)
            if not file_path or not file_path.exists():
                continue
            self._enriched_attempted.add((indexer.group_id, entry.id))

            async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
                await self._on_enriched(_indexer, file_id, fields)

            self._photo_enricher.spawn(entry, file_path, on_done)

    async def _enrich_photo_roots_now(self, group_id: str) -> None:
        """
        Photos app: sweep a group's existing index right after its
        photo_roots set changes (ops.set_photo_roots). Mirrors
        `_enrich_video_root_now`/`_enrich_audio_root_now` — the ordinary
        path above only ever looks at entries new since the last broadcast,
        so a folder that already had photos in it before it was added to
        photo_roots would otherwise never get enriched at all. Also covers
        a root being *removed*: nothing un-enriches on removal (the cache
        entry is harmless, just unused — docs/photos.md's cache is
        disposable), so re-sweeping the new set is enough.
        """
        indexer = self._state.get("indexers", {}).get(group_id)
        if not indexer:
            return
        await self._enrich_new_photo_entries(indexer, list(indexer.index.entries))

    async def _reenrich_renamed_photo_entries(
        self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex,
    ) -> None:
        """
        Photos app equivalent of `_reenrich_renamed_video_entries` — a
        rename changes nothing about the image's own bytes (thumbnail, EXIF
        fields are content-derived, not name-derived), so this exists only
        for consistency/symmetry with Videos/Music and to catch the case of
        a file moving *into* a newly-covered photo_roots subtree via a
        rename rather than a fresh add. Re-running enrichment on an
        unchanged file is redundant work, not a correctness issue.
        """
        for entry in updates:
            if entry.type != "image":
                continue
            old = previous.get_entry(entry.id)
            if old is None or (old.name == entry.name and old.path == entry.path):
                continue
            self._enriched_attempted.discard((indexer.group_id, entry.id))
        await self._enrich_new_photo_entries(indexer, updates)

    async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None:
        """
        Merge enrichment fields into the live index and re-trigger a
        broadcast so they reach connected clients as an INDEX_DELTA update
        (GroupIndex.diff's `updates`, not `additions` — same id, new fields).

        Builds a *new* IndexEntry via dataclasses.replace rather than
        mutating the existing one in place: the diff mechanism compares
        against a shallow snapshot of entry *references*, so an in-place
        mutation would silently also change what "previous" looks like,
        and the change would never show up as a diff (see group_index.py's
        diff() docstring).
        """
        idx = indexer.index
        entry = idx.get_entry(file_id)
        if entry is None:
            return   # removed from the index while enrichment was in flight
        idx.add_entry(replace(entry, **fields))
        await self._on_index_change(indexer)

    def _drop_group_sessions(self, group_id: str) -> None:
        """Close live sessions for a revoked group (H4)."""
        if not self._webrtc or not group_id:
            return
        for session in list(self._webrtc._sessions.values()):
            if session._group_id == group_id:
                asyncio.ensure_future(session.close())
                log.info("Dropped session for revoked group %s", group_id[:8])

    async def _register_swarm(self, hashes: list[str], endpoint: str) -> None:
        try:
            n = await self._hub.register_swarm(hashes, endpoint)
            log.info("Swarm: registered %d/%d hashes", n, len(hashes))
        except Exception as e:
            log.warning("Swarm registration failed: %s", e)

    async def _shutdown(self) -> None:
        log.info("Shutting down...")
        self._state["status"] = "stopping"

        for handle in self._pending_broadcasts.values():
            handle.cancel()
        self._pending_broadcasts.clear()

        for task in self._tasks:
            task.cancel()
        for task in self._tasks:
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

        if self._webrtc:
            await self._webrtc.close_all()

        if self._audit_store:
            await self._audit_store.close()

        if self._bundle_store:
            await self._bundle_store.close()

        if self._tmdb_client:
            await self._tmdb_client.close()

        if self._musicbrainz_client:
            await self._musicbrainz_client.close()

        if self._media_cache:
            await self._media_cache.close()

        if self._roster:
            await self._roster.close()

        for store in self._chat_stores.values():
            await store.close()

        if self._index_cache:
            await self._index_cache.close()

        for indexer in self._indexers:
            await indexer.stop()

        if self._quic_server:
            await self._quic_server.stop()

        token_file = getattr(self, "_ui_token_file", None)
        if token_file is not None:
            token_file.unlink(missing_ok=True)

        log.info("Node stopped")


# ── CLI helpers ───────────────────────────────────────────────────────────────

def _daemon_api(cfg: Config, path: str, method: str = "GET",
                timeout: int = 30, body: dict | None = None) -> dict:
    """
    Call the daemon's loopback API.

    The daemon owns the roster, the hub session and the live group contexts, so
    the CLI asks it to act rather than opening its databases behind its back. It
    also means every operator action goes through the same authorization as the
    admin UI (the per-run session token, 11.5.3).
    """
    import json as _json
    import urllib.error
    import urllib.parse
    import urllib.request

    token_file = cfg.data_dir / "ui-token"
    if not token_file.exists():
        print("Node is not running — start it with: meshbay-node")
        sys.exit(1)

    sep = "&" if "?" in path else "?"
    url = (f"http://127.0.0.1:{cfg.node.ui_port}{path}"
           f"{sep}t={token_file.read_text().strip()}")
    try:
        data = _json.dumps(body).encode() if body is not None else None
        req = urllib.request.Request(
            url, method=method, data=data,
            headers={"Content-Type": "application/json"} if data else {})
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return _json.loads(r.read())
    except urllib.error.HTTPError as e:
        raw = e.read().decode()[:600]
        try:
            parsed = _json.loads(raw)
            detail = parsed.get("error", raw)
            # Endpoints that refuse a name offer the ones that would work; a bare
            # "no such group" leaves the operator guessing at a UUID.
            for row in parsed.get("available", []):
                detail += f"\n  {row.get('name', ''):<24} {row.get('id', '')}"
        except Exception:
            detail = raw
        print(f"failed: {detail}")
        sys.exit(1)
    except Exception as e:
        print(f"failed: {e}")
        sys.exit(1)


def _resolve_group(cfg: Config, group: str | None) -> str:
    """
    The group argument as an id, or the only configured one.

    Accepts a name as well, because node.toml already gives every group one and
    nobody remembers a UUID. A name that matches nothing configured says so, and
    lists what is — silently passing it through produced a 404 from the daemon
    that read like the group did not exist on the hub.
    """
    if group:
        by_id = [g for g in cfg.groups if g.id == group]
        if by_id:
            return by_id[0].id
        by_name = [g for g in cfg.groups if g.name == group and g.id]
        if len(by_name) == 1:
            return by_name[0].id
        if len(by_name) > 1:
            print(f"several groups in node.toml are named {group!r} — use the id")
            sys.exit(1)
        # An id this node does not host is still worth passing on: the daemon
        # gives the better error, naming the group it does host.
        if "-" in group and len(group) == 36:
            return group
        print(f"no group named {group!r} in {DEFAULT_CONFIG_PATH}")
        if cfg.groups:
            print("configured groups:")
            for g in cfg.groups:
                print(f"  {g.name or '(unnamed)':<24} {g.id or '(no id yet)'}")
        sys.exit(1)
    configured = [g.id for g in cfg.groups if g.id]
    if len(configured) == 1:
        return configured[0]
    print("--group is required (several groups configured)"
          if configured else "no group configured in node.toml")
    sys.exit(1)


def _systemctl_user(verb: str, unit: str, *, not_running_hint: str,
                    success: str, watch: str | None) -> None:
    """
    Run `systemctl --user <verb> <unit>` and report the result.

    The lifecycle authority is the unit, not this process: systemd already
    knows which PID it started, restarts it on failure (`Restart=on-failure`
    in the unit) and reloads it correctly (`ExecReload=`). Anything this CLI
    did instead — finding a process by pattern-matching its command line,
    signalling it, respawning it — is a second, worse implementation of what
    systemd is already doing, and pattern-matching a process list has already
    hit a real developer's real running node by accident.
    """
    import subprocess

    result = subprocess.run(["systemctl", "--user", verb, unit],
                            capture_output=True, text=True)
    if result.returncode != 0:
        detail = (result.stderr or result.stdout).strip()
        print(detail or not_running_hint)
        sys.exit(1)
    print(success)
    if watch:
        print(watch)


# ── Entry point ───────────────────────────────────────────────────────────────

def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description="MeshBay Node daemon")
    parser.add_argument("command", nargs="?",
                        choices=["init", "reset", "status", "ui", "gek-init",
                                 "gek", "operator", "member", "group", "file",
                                 "video", "denylist", "stun", "reload",
                                 "restart-daemon", "calibrate-argon2"],
                        help="init: provision config + keystore | reset: erase all "
                             "node state | status: node state and keys "
                             "| ui: print the admin UI URL | operator pair: pair a "
                             "browser with this node | member list|invite|revoke|unpin "
                             "| group list|add|remove | gek init|rotate | file list|rm "
                             "| video rematch: re-resolve TMDB matches for a group's "
                             "videos | denylist show|clear "
                             "| stun list|add|remove|reset "
                             "| reload: re-read node.toml "
                             "(systemctl --user reload) | restart-daemon: restart "
                             "the systemd unit (systemctl --user restart) "
                             "| calibrate-argon2: benchmark")
    parser.add_argument("subcommand", nargs="?",
                        help="'pair' for operator; list|invite|revoke|unpin for "
                             "member; list|add|remove for group; init|rotate for gek; "
                             "list|rm for file; rematch for video; show|clear for "
                             "denylist; list|add|remove|reset for stun")
    parser.add_argument("target", nargs="?",
                        help="username for member invite|revoke|unpin; group name "
                             "for group add; file id for file rm; identifier for "
                             "denylist clear")
    parser.add_argument("--hub-url", default=None,
                        help="hub URL, for init (e.g. https://meshbay.org)")
    parser.add_argument("--username", default=None,
                        help="hub username, for init")
    parser.add_argument("--dir", default=None,
                        help="shared directory, for group add")
    parser.add_argument("--upload-dir", default=None,
                        help="separate upload directory, for group add")
    parser.add_argument("--yes", action="store_true",
                        help="skip the confirmation for destructive commands")
    parser.add_argument("--config", type=Path, default=None,
                        help="Config file path")
    parser.add_argument("--group", default=None,
                        help="group id (optional if only one is configured)")
    parser.add_argument("--log-level", default="INFO",
                        choices=["DEBUG", "INFO", "WARNING", "ERROR"])
    args = parser.parse_args()

    # Query commands print a report; library logging would interleave with it.
    quiet = args.command in ("status", "ui", "gek-init", "gek", "operator",
                             "member", "group", "file", "video", "denylist",
                             "stun", "reload", "restart-daemon", "reset")
    logging.basicConfig(
        level=logging.ERROR if quiet else getattr(logging, args.log_level),
        format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
    )

    if args.command == "init":
        cfg_path = args.config or DEFAULT_CONFIG_PATH
        config_dir = cfg_path.parent
        config_dir.mkdir(parents=True, exist_ok=True)

        hub_url = args.hub_url
        username = args.username

        if not hub_url:
            hub_url = input("Hub URL [https://meshbay.org]: ").strip() or "https://meshbay.org"
        if not username:
            username = input("Hub username: ").strip()
        if not username:
            print("Username is required.")
            sys.exit(1)

        if cfg_path.exists():
            existing = cfg_path.read_text()
            import re as _re
            m = _re.search(r'username\s*=\s*"([^"]*)"', existing)
            existing_user = m.group(1) if m else ""
            if existing_user and existing_user != "myusername" and existing_user != username:
                print(f"Config already exists with username {existing_user!r}.")
                print("This node belongs to another operator. Use 'meshbay-node reset' first.")
                sys.exit(1)
            if existing_user in ("", "myusername"):
                updated = _re.sub(
                    r'(username\s*=\s*)"[^"]*"', rf'\1"{username}"', existing)
                updated = _re.sub(
                    r'(url\s*=\s*)"[^"]*"', rf'\1"{hub_url}"', updated, count=1)
                cfg_path.write_text(updated)
                print(f"Config updated: username={username}, hub={hub_url}")
            else:
                print(f"Config already exists: {cfg_path}")
        else:
            unlock_file = config_dir / "unlock.key"
            toml_lines = [
                "[hub]",
                f'url      = "{hub_url}"',
                f'username = "{username}"',
                "",
                "[node]",
                "quic_port = 19010",
                "ui_port   = 18000",
                "",
                "[keystore]",
                f'unlock_file = "{unlock_file}"',
                "",
            ]
            cfg_path.write_text("\n".join(toml_lines) + "\n")
            os.chmod(cfg_path, 0o600)
            print(f"Config written to {cfg_path}")

        unlock_file = config_dir / "unlock.key"
        if not unlock_file.exists():
            import secrets
            key = secrets.token_urlsafe(32)
            unlock_file.write_text(key + "\n")
            os.chmod(unlock_file, 0o600)
            print(f"Unlock key created: {unlock_file}")

        cfg = load_config(cfg_path)
        if cfg.keystore.path.exists():
            print(f"Keystore already exists: {cfg.keystore.path}")
            keys = load_keystore(
                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
        else:
            keys = create_keystore(
                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
            print(f"Keystore created: {cfg.keystore.path}")

        print(f"Node key: {keys.pk_ed25519_b64}")
        print()
        print("Next steps:")
        print(f"  1. Link this node key on {hub_url} → Settings → Link Node")
        print("  2. systemctl --user enable --now meshbay-node")
        print("  3. meshbay-node group add <name> --dir /path/to/files")
        print("  4. meshbay-node gek init")
        print("  5. meshbay-node operator pair")
        return

    if args.command == "reset":
        import shutil

        config_dir = Path.home() / ".config" / "meshbay"
        data_dir = Path.home() / ".local" / "share" / "meshbay"
        state_dir = Path.home() / ".local" / "state" / "meshbay"

        items = []
        for d in (config_dir, data_dir):
            if d.exists():
                for child in sorted(d.iterdir()):
                    items.append(child)

        if not items:
            print("Nothing to reset — no node state found.")
            return

        print("This will permanently erase all node state:")
        for p in items:
            print(f"  {p}")
        print()
        print("WARNING: a new keystore means a new identity. All group")
        print("memberships, operator pairings, and invitations are lost.")

        if not args.yes:
            answer = input("\nProceed? [y/N] ").strip().lower()
            if answer != "y":
                print("Aborted.")
                return

        import subprocess as _sp
        import json as _json
        import urllib.request
        import urllib.error

        token_file = data_dir / "ui-token"
        if token_file.exists():
            try:
                cfg = Config(config_dir / "node.toml")
                tok = token_file.read_text().strip()
                url = (f"http://127.0.0.1:{cfg.node.ui_port}"
                       f"/api/unlink?t={tok}")
                req = urllib.request.Request(url, method="DELETE")
                with urllib.request.urlopen(req, timeout=5) as r:
                    _json.loads(r.read())
                print("Unlinked node key from hub.")
            except Exception:
                print("Could not unlink from hub (daemon not reachable).")

        _sp.run(["systemctl", "--user", "disable", "--now", "meshbay-node"],
                capture_output=True)

        for d in (config_dir, data_dir):
            if d.exists():
                shutil.rmtree(d)
                print(f"Removed {d}")
        if state_dir.is_dir():
            shutil.rmtree(state_dir)
            print(f"Removed {state_dir}")
        print("Node state erased. Run 'meshbay-node init' to start over.")
        return

    if args.command == "calibrate-argon2":
        calibrate_argon2()
        return

    if args.command == "status":
        import json as _json
        import urllib.request

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        print(f"hub          {cfg.hub.url}  (user {cfg.hub.username or '—'})")

        try:
            keys = load_keystore(
                path=cfg.keystore.path, unlock_file=cfg.keystore.unlock_file)
            print(f"node key     {keys.pk_ed25519_b64}")
        except FileNotFoundError:
            print("node key     <no keystore — run: meshbay-node init>")
        except Exception as e:
            print(f"node key     <keystore locked: {e}>")

        token_file = cfg.data_dir / "ui-token"
        live = None
        if token_file.exists():
            try:
                url = (f"http://127.0.0.1:{cfg.node.ui_port}"
                       f"/api/status?t={token_file.read_text().strip()}")
                with urllib.request.urlopen(url, timeout=3) as r:
                    live = _json.loads(r.read())
            except Exception:
                live = None

        if live:
            print(f"daemon       running — {live.get('status')}")
            print(f"node_id      {live.get('endpoint_hint') or '—'}")
            print(f"groups       {live.get('group_count', 0)}"
                  f"  files {live.get('total_files', 0)}"
                  f"  peers {live.get('webrtc_peers', 0)}")
            print(f"admin UI     meshbay-node ui")

            needs = live.get("needs", [])
            if needs:
                _GUIDANCE = {
                    "node_key_link": (
                        "Link node key",
                        f"Copy the node key above and paste it in Settings → Link Node on {cfg.hub.url}"),
                    "group_add": (
                        "Add a group",
                        "meshbay-node group add <name> --dir /path/to/files"),
                    "operator_pair": (
                        "Pair as operator",
                        "meshbay-node operator pair"),
                }
                print()
                print("action needed:")
                for need in needs:
                    if need.startswith("gek_init:"):
                        name = need.split(":", 1)[1]
                        print(f"  → Initialize group key for {name}")
                        print(f"    meshbay-node gek init --group \"{name}\"")
                    elif need in _GUIDANCE:
                        label, hint = _GUIDANCE[need]
                        print(f"  → {label}")
                        print(f"    {hint}")
                    else:
                        print(f"  → {need}")
        else:
            print("daemon       not running")

        print(f"config       {DEFAULT_CONFIG_PATH}")
        if not cfg.groups:
            print("groups       none configured — create a group on the hub, then add")
            print("             a [[groups]] entry with its id and a directory")
        else:
            for g in cfg.groups:
                print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
                if not g.roots:
                    print("             <no directory configured>")
                for r in g.roots:
                    label = r.name or Path(r.path).name
                    flag = "  (uploads)" if r.upload else ""
                    live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
                    print(f"             {label} → {r.path}{flag}{live}")
        # Node authority: the roster is the source of truth, node.toml the legacy
        # form. Read the DB directly so this reports correctly while the daemon is
        # stopped — the state an operator is most often in when checking.
        import asyncio as _asyncio

        from meshbay_node.roster import Roster as _Roster

        async def _read_roster() -> tuple[list, int]:
            r = _Roster(db_path=cfg.data_dir / "roster.db")
            await r.open()
            try:
                return (await r.list_members()), len(await r.list_invites())
            finally:
                await r.close()

        try:
            members, pending = _asyncio.run(_read_roster())
        except Exception as e:
            members, pending = [], 0
            print(f"roster       <unreadable: {e}>")

        operators = [m for m in members if m["role"] == "operator"
                     and m["status"] == "active"]
        if operators:
            for op in operators:
                print(f"operator     {op.get('username') or op['user_id'][:8]}"
                      f"  key {(op.get('pk_ed25519') or '')[:16]}…"
                      f"  paired {op.get('pinned_at', '?')}")
        else:
            print("operator     NONE PAIRED — file deletion and member invites are")
            print("             refused. Run: meshbay-node operator pair")
        if pending:
            print(f"invites      {pending} pending code(s)")
        return

    if args.command == "member":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "list"

        if sub == "list":
            group = args.group or ""
            out = _daemon_api(
                cfg, f"/api/roster?group_id={group}" if group else "/api/roster")
            identities = {i["user_id"]: i for i in out.get("identities", [])}

            members = out.get("members", [])
            if not members:
                print("no members admitted yet")
                print("invite someone:  meshbay-node member invite <username>")
            for m in members:
                ident = identities.get(m["user_id"], {})
                scope = m["group_id"][:8] if m["group_id"] else "node-wide"
                print(f"{(ident.get('username') or m['user_id'])[:20]:20} "
                      f"{m['role']:9} {m['status']:8} {scope:10} "
                      f"pinned {ident.get('pinned_at', '?')} "
                      f"({ident.get('pinned_via', '?')})")

            invites = out.get("invites", [])
            if invites:
                print()
                for i in invites:
                    print(f"pending invite  user {i['user_id'][:12]}  "
                          f"group {(i['group_id'] or 'node-wide')[:8]}  "
                          f"expires {i['expires_at']}")
            return

        if not args.target:
            print(f"usage: meshbay-node member {sub} <username>")
            sys.exit(1)

        if sub == "invite":
            group_id = _resolve_group(cfg, args.group)
            out = _daemon_api(
                cfg, f"/api/groups/{group_id}/invites?username={args.target}",
                method="POST")
            from meshbay_node.roster import write_code_file
            path = write_code_file(cfg.data_dir, out["code"],
                                   out.get("expires_at", ""), name="invite-code")
            print(f"INVITATION CODE  {out['code']}")
            print(f"valid until      {out.get('expires_at', '?')}")
            print()
            print(f"Send it to {args.target} however you normally talk. It works")
            print("once, for that account only, and never passes through the hub.")
            print("They enter it the first time they open the group — you do not")
            print("need to be online then.")
            print()
            print(f"also written to {path}")
            return

        # revoke and unpin both name a person; the daemon resolves the account.
        # It tries its own roster first and falls back to the hub, so a node that
        # pinned someone before invitations carried a name is still manageable.
        match = _daemon_api(cfg, f"/api/resolve?username={args.target}")

        if sub == "revoke":
            group_id = _resolve_group(cfg, args.group)
            out = _daemon_api(
                cfg, f"/api/members/{match['user_id']}/revoke?group_id={group_id}",
                method="POST")
            print(f"{args.target} revoked from {group_id[:8]}")
            print("They stop receiving the group key on their next connection.")
            print("They still hold the current one — rotate it:")
            print(f"  meshbay-node gek-init --group {group_id}")
            return

        if sub == "unpin":
            _daemon_api(cfg, f"/api/members/{match['user_id']}/unpin", method="POST")
            print(f"{args.target} unpinned — they can pair again with a new key")
            print(f"issue a code:  meshbay-node member invite {args.target}")
            return

        print("usage: meshbay-node member list|invite|revoke|unpin")
        sys.exit(1)

    if args.command in ("gek-init", "gek"):
        # `gek-init` is the original spelling and still works. `gek rotate` is
        # the one that matters after a revocation: the ex-member holds the
        # current key and nothing else takes it from them.
        sub = "init" if args.command == "gek-init" else (args.subcommand or "init")
        if sub not in ("init", "rotate"):
            print("usage: meshbay-node gek init|rotate [--group NAME]")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        group_id = _resolve_group(cfg, args.group)

        if sub == "rotate" and not args.yes:
            print("Rotating replaces this group's key.")
            print("  · every member re-receives it automatically on their next connect")
            print("  · anyone revoked keeps the OLD key and loses access to new content")
            print("  · content already downloaded stays readable to whoever has it")
            if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"):
                print("cancelled")
                return

        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek"
                               f"{'?rotate=true' if sub == 'rotate' else ''}",
                          method="POST", timeout=60)

        verb = "rotated" if out.get("rotated") else "ready"
        print(f"GEK {verb} for {group_id}")
        print(f"  {out.get('authorized_members', 0)} authorized member(s) — each "
              f"receives the key on connect")
        for err in out.get("errors") or []:
            print(f"  ! {err}")
        return

    if args.command == "reload":
        # Milestone 14.8. The daemon re-reads node.toml; groups that appeared or
        # whose roots changed are picked up without dropping live connections.
        #
        # Delegated to systemd rather than hunting a PID with pgrep and signalling
        # it directly: an unanchored (or merely unlucky) pattern match there has
        # already SIGHUPed a developer's own running node by accident — see the
        # comment this replaced, and test_cli_dispatch.py's stub_daemon fixture,
        # which had to stub os.kill for exactly that reason. The unit already
        # declares `ExecReload=/bin/kill -HUP $MAINPID`, so systemd sends the
        # signal to the one process it actually started.
        _systemctl_user(
            "reload", "meshbay-node",
            not_running_hint="Node is not running as a systemd unit — start it "
                              "with: systemctl --user start meshbay-node",
            success="sent reload to meshbay-node",
            watch="watch the result:  journalctl --user -u meshbay-node -f")
        return

    if args.command == "restart-daemon":
        # Same reasoning as reload: no PID hunting, no manual respawn — systemd
        # already knows how to stop and start this unit, and does not need this
        # process to guess where its log file is.
        _systemctl_user(
            "restart", "meshbay-node",
            not_running_hint="meshbay-node is not installed as a systemd unit — "
                              "see packaging/systemd/",
            success="meshbay-node restarted via systemd",
            watch="check status:  systemctl --user status meshbay-node\n"
                  "watch logs:    journalctl --user -u meshbay-node -f")
        return

    if args.command == "denylist":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "show"

        if sub == "show":
            out = _daemon_api(cfg, "/api/denylist")
            total = out.get("count", 0)
            if not total:
                print("denylist     empty — nothing is being refused")
                return
            for kind in ("users", "groups", "jtis"):
                for entry in out.get(kind, []):
                    print(f"  {kind[:-1]:<6} {entry}")
            print(f"\n{total} entr(y/ies). These survive a restart (finding H4).")
            return

        if sub == "clear":
            if not args.yes:
                what = args.target or "EVERY entry"
                print(f"Clearing the denylist re-admits {what}.")
                print("A revocation the hub sent will not come back on its own.")
                if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}",
                              method="POST")
            print(f"removed {out['removed']} entr(y/ies) ({out['subject']})")
            return

        print("usage: meshbay-node denylist show|clear [identifier] [--yes]")
        sys.exit(1)

    if args.command == "stun":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "list"

        if sub == "list":
            out = _daemon_api(cfg, "/api/node-settings")
            servers = out.get("stun_servers", [])
            if not servers:
                print("stun servers     (none configured)")
                return
            for i, s in enumerate(servers, 1):
                print(f"  {i}. {s}")
            return

        if sub == "add":
            url = args.target
            if not url:
                print("usage: meshbay-node stun add <stun:host:port>")
                sys.exit(1)
            if not url.startswith("stun:"):
                print(f"error: STUN URL must start with stun: — got {url!r}")
                sys.exit(1)
            out = _daemon_api(cfg, "/api/node-settings")
            servers = out.get("stun_servers", [])
            if url in servers:
                print(f"already present: {url}")
                return
            servers.append(url)
            _daemon_api(cfg, "/api/node-settings", method="PUT",
                        body={"stun_servers": servers})
            print(f"added {url} ({len(servers)} servers total)")
            return

        if sub == "remove":
            url = args.target
            if not url:
                print("usage: meshbay-node stun remove <stun:host:port>")
                sys.exit(1)
            out = _daemon_api(cfg, "/api/node-settings")
            servers = out.get("stun_servers", [])
            if url not in servers:
                print(f"not found: {url}")
                sys.exit(1)
            servers.remove(url)
            _daemon_api(cfg, "/api/node-settings", method="PUT",
                        body={"stun_servers": servers})
            print(f"removed {url} ({len(servers)} servers remaining)")
            return

        if sub == "reset":
            from meshbay_node.config import DEFAULT_STUN_SERVERS
            _daemon_api(cfg, "/api/node-settings", method="PUT",
                        body={"stun_servers": list(DEFAULT_STUN_SERVERS)})
            print("STUN servers reset to defaults:")
            for s in DEFAULT_STUN_SERVERS:
                print(f"  {s}")
            return

        print("usage: meshbay-node stun list|add|remove|reset [url]")
        sys.exit(1)

    if args.command == "file":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "list"
        group_id = _resolve_group(cfg, args.group)

        if sub == "list":
            out = _daemon_api(cfg, f"/api/groups/{group_id}/files")
            files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"]))
            if not files:
                print("no files indexed")
                return
            for f in files:
                print(f"  {f['id'][:12]}  {f['size']:>12}  {f['path']}/{f['name']}")
            print(f"\n{len(files)} file(s). Remove one with: "
                  f"meshbay-node file rm <id>")
            return

        if sub == "rm":
            # Milestone 14.11 — the last operator action that needed a browser.
            if not args.target:
                print("usage: meshbay-node file rm <file-id> [--group NAME]")
                sys.exit(1)
            out = _daemon_api(cfg, f"/api/groups/{group_id}/files",)
            matches = [f for f in out.get("files", [])
                       if f["id"].startswith(args.target)]
            if not matches:
                print(f"no file whose id starts with {args.target!r}")
                sys.exit(1)
            if len(matches) > 1:
                print(f"{args.target!r} matches {len(matches)} files — be more specific:")
                for f in matches[:10]:
                    print(f"  {f['id'][:16]}  {f['path']}/{f['name']}")
                sys.exit(1)
            target = matches[0]
            if not args.yes:
                print(f"Delete {target['path']}/{target['name']} "
                      f"({target['size']} bytes) from disk?")
                print("This removes the file itself, not just the listing.")
                if input("delete? [y/N] ").strip().lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}",
                        method="DELETE")
            print(f"deleted {target['path']}/{target['name']}")
            return

        print("usage: meshbay-node file list|rm <id> [--group NAME] [--yes]")
        sys.exit(1)

    if args.command == "video":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        if (args.subcommand or "") != "rematch":
            print("usage: meshbay-node video rematch [--group NAME] [--yes]")
            sys.exit(1)
        group_id = _resolve_group(cfg, args.group)
        if not args.yes:
            print("Re-resolve every automatic TMDB match for this group's videos?")
            print("Manual 'Fix match' corrections are kept. Re-resolution is lazy —")
            print("each poster re-queries TMDB the next time it is opened.")
            if input("proceed? [y/N] ").strip().lower() not in ("y", "yes"):
                print("cancelled")
                return
        out = _daemon_api(cfg, f"/api/groups/{group_id}/video/rematch", method="POST")
        print(f"cleared {out.get('removed', 0)} automatic match(es) "
              f"across {out.get('videos', 0)} video file(s)")
        return

    if args.command == "group":
        if args.subcommand in (None, "list"):
            # Milestone 14.2.
            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
            out = _daemon_api(cfg, "/api/groups")
            groups = out.get("groups", [])
            if not groups:
                print("no groups hosted — add one with: "
                      "meshbay-node group add <name> --dir <path>")
                return
            for g in groups:
                key = "GEK" if g.get("has_gek") else "NO KEY"
                print(f"  {g['name']}  [{g['visibility']}/{g.get('join_policy')}]  "
                      f"{key}  {g['file_count']} file(s)  "
                      f"{g.get('peers', 0)} peer(s)")
                print(f"      {g['id']}")
                for r in g.get("roots", []):
                    flags = ""
                    if r.get("upload"):
                        flags = " (uploads, direct)" if r.get("direct") else " (uploads)"
                    live = "" if r.get("available", True) else "  [UNAVAILABLE]"
                    print(f"      root {r['name']}{flags}{live}")
                if not g.get("has_gek"):
                    print(f"      give it a key:  meshbay-node gek init "
                          f"--group {g['name']}")
            return

        if args.subcommand == "remove":
            if not args.target:
                print("usage: meshbay-node group remove <name>")
                sys.exit(1)
            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
            if not args.yes:
                answer = input(f"Remove group '{args.target}' from this node? [y/N] ")
                if answer.lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            out = _daemon_api(cfg, "/api/groups/detach", method="POST",
                              body={"name": args.target})
            print(f"{out['name']} ({out['group_id'][:8]}) removed from {out['config']}")
            print()
            print("Restart the daemon to stop hosting it:")
            print("  meshbay-node restart-daemon")
            return

        if args.subcommand != "add":
            print("usage: meshbay-node group list|add|remove <name>")
            sys.exit(1)
        if not args.target or not args.dir:
            print("usage: meshbay-node group add <name> --dir <path> [--upload-dir <path>]")
            print()
            print("The group must already exist on the hub and be yours. This")
            print("only tells the node to host it, and picks the directory.")
            print("--upload-dir sets a separate directory for uploaded files.")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        body = {"name": args.target, "shared_dir": args.dir}
        if args.upload_dir:
            body["upload_dir"] = args.upload_dir
        out = _daemon_api(cfg, "/api/groups/attach", method="POST", body=body)
        print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
        print(f"  shared_dir   {out['shared_dir']}")
        if out.get("upload_dir"):
            print(f"  upload_dir   {out['upload_dir']}")
        print()
        print("Tell the daemon to re-read its config, then give the group a key:")
        print("  meshbay-node reload")
        print(f"  meshbay-node gek init --group {out['name']}")
        print()
        print("The key is this group's own — members of your other groups cannot")
        print("read it, and joining one says nothing about the other.")
        return

    if args.command == "operator":
        if args.subcommand != "pair":
            print("usage: meshbay-node operator pair")
            sys.exit(1)
        if args.group:
            # Silently ignoring it invited the reading that a code belongs to a
            # group, and then that pairing had not worked because the group did
            # not change.
            print("operator pair takes no --group: pairing is node-wide.")
            print("One paired browser can invite to, and delete files in, every")
            print("group this node hosts.")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        out = _daemon_api(cfg, "/api/operator/pair", method="POST")

        from meshbay_node.roster import write_code_file
        path = write_code_file(cfg.data_dir, out["code"], out.get("expires_at", ""))

        print(f"PAIRING CODE   {out['code']}")
        print(f"valid until    {out.get('expires_at', '?')}")
        print()
        print("Sign in to the web app as this node's operator, open one of your")
        print("groups, go to the Members tab and enter the code there.")
        print("It works once, for that account only, and authorizes invites and")
        print("file deletion from that browser.")
        print()
        print(f"also written to {path}")
        return

    if args.command == "ui":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        token_file = cfg.data_dir / "ui-token"
        if not token_file.exists():
            print("Node does not appear to be running — start it with: meshbay-node")
            sys.exit(1)
        print(f"http://127.0.0.1:{cfg.node.ui_port}"
              f"/?t={token_file.read_text().strip()}")
        print()
        print("The UI listens on loopback only. From another machine:")
        print(f"  ssh -L {cfg.node.ui_port}:127.0.0.1:{cfg.node.ui_port} <this-host>")
        return

    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
    if not cfg.hub.username:
        print("Error: hub.username not set in config. Run: meshbay-node init")
        sys.exit(1)

    daemon = NodeDaemon(cfg, Path(args.config or DEFAULT_CONFIG_PATH))
    asyncio.run(daemon.run())


if __name__ == "__main__":
    main()