path: root/packages/meshbay-node/src/meshbay_node/daemon.py

"""
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 control API on node.ui_port (loopback only, token-gated)
  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 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
import time
from pathlib import Path

import uvicorn

from meshbay_common.background import spawn
from meshbay_common.paths import fold
from meshbay_common import MNP_VERSION
from meshbay_common.protocol import MNP
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 import uploads as uploads_mod
from meshbay_node.musicbrainz import MusicBrainzClient
from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore
from meshbay_node.platform import chmod_private, config_dir, data_dir, state_dir
from meshbay_node.roster import Roster
from meshbay_node.transport import (
    Denylist,
    QUIC_AVAILABLE,
    WEBRTC_AVAILABLE,
)
from meshbay_node.transport.wire import index_delta_message, index_sync_message

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

log = logging.getLogger(__name__)


def _under_any_directory(path: str, directories: list[str]) -> bool:
    """
    Whether an entry's folder is one of an app's directories, or inside one.

    Mirrors `underAnyDirectory` in the SPA's app modules. One helper for every
    app since they all take a list: Videos and Music used to take a single
    folder and had a function each saying the same thing, which is how the two
    came to differ in what they did with a trailing slash.
    """
    path = path or ""
    return any(path == d or path.startswith(d + "/") for d in directories)


def _owning_directory(path: str, directories: list[str]) -> str | None:
    """
    Which of an app's directories an entry belongs to — the deepest match.

    Deepest, because directories may nest: with both `Media` and
    `Media/Albums` configured, a file under the second belongs to the second.
    Taking the first match instead would measure it against a boundary one
    level too shallow, which for Music is the difference between reading a
    folder as an artist and reading it as a release.
    """
    path = path or ""
    best: str | None = None
    for d in directories:
        if path == d or path.startswith(d + "/"):
            if best is None or len(d) > len(best):
                best = d
    return best


# ── 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 ────────────────────────────────────────────────────────────────────

def _root_shape(roots) -> set[tuple]:
    """What has to match for a group's roots to count as unchanged on reload."""
    return {(r.name, str(r.path), r.writable, r.removable) for r in roots}


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 the local control API early (so the operator can read the
        # node key before hub login). It is JSON-only, loopback-only, and both
        # the CLI and the desktop client's Node page are its clients.
        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 control API (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 the CLI and the desktop client can read it — nobody
        # should ever copy a token out of a log or a terminal.
        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, encoding="utf-8", newline="\n")
        chmod_private(self._ui_token_file)
        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("Control API on 127.0.0.1:%d", self._config.node.ui_port)

        self._tasks.append(asyncio.create_task(self._reap_partial_uploads()))

        # 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 = await self._build_roots(group_cfg)
                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 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,
                    on_root_ejected=self._eject_persister(group_cfg.id),
                    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,
                    # Bound method, called when an upload finishes. The entry
                    # it belongs to does not exist yet (see
                    # webrtc_server._register_uploader), so the indexer keeps
                    # the record and stamps the entry when it creates it.
                    "record_upload": indexer.record_upload,
                    # Shown to the operator in Settings, and kept current in
                    # place by set_scan_settings (ops.py) — same reasoning as
                    # enabled_apps 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,
                    # Read once at load, kept current in place by the signed
                    # operation that changes it — the upload handler is
                    # synchronous and a database round trip per chunk would be
                    # absurd. (Whether a member may upload is not here any
                    # more: it is `writable` on the root being written to,
                    # which the RootSet above already carries.)
                    "enabled_apps": await self._roster.enabled_apps(
                        group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS),
                    # How many transfers one member may run at once here. Empty
                    # means the operator has not said, and the node's default
                    # applies — never "unlimited" (transfers.member_cap).
                    "transfer_limits": (
                        await self._roster.transfer_limits(group_cfg.id)
                        if self._roster else {}),
                    # Which folder(s) inside the shared roots each app works
                    # over. One shape for every app (roster.py's
                    # app_directories) — an empty list means nothing has been
                    # chosen, which every app reads as "show nothing yet",
                    # never "the whole group index".
                    **(await self._app_directories_ctx(group_cfg.id)),
                    # Whether the node unfurls links members post here.
                    "chat_link_preview": await self._roster.chat_link_preview(
                        group_cfg.id) if self._roster else True,
                    # Which chat epoch key is current. Opened here if the
                    # group has none, because chat is always encrypted (MNP
                    # 2.0) and a group with no epoch is a group nobody can
                    # speak in — the node cannot wait for an operator to notice.
                    "chat_epoch": await self._ensure_chat_epoch(group_cfg.id),
                    # 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 the app directories 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 — the control API 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,
                    max_concurrent_downloads=self._config.node.max_concurrent_downloads,
                    max_concurrent_uploads=self._config.node.max_concurrent_uploads,
                    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)
            #
            # Off unless `[node] quic_enabled = true`: no shipping client speaks
            # QUIC (browser and desktop use WebRTC; the `group://` sidecar is
            # unbuilt), so starting it by default only exposes a UDP port.
            if QUIC_AVAILABLE and self._config.node.quic_enabled:
                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))
            elif QUIC_AVAILABLE:
                log.info("QUIC server disabled ([node] quic_enabled = false)")

            # 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 control API state (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
            # The node's own Ed25519 key. Needed by `encrypt_chat_history`,
            # which seals migrated messages under a synthetic device of the
            # node's rather than pretending to hold a member's signing key.
            self._state["sk_node"] = keys.sk_ed25519
            self._state["webrtc"] = self._webrtc
            self._state["quic_server"] = self._quic_server
            self._state["hub"] = hub
            self._state["reload_fn"] = self._reload_config
            # Keyed by app, so `ops.set_app_directories` finds the right
            # sweep without knowing which apps exist — an app with nothing to
            # enrich simply has no entry.
            self._state["enrich_app_dirs_fns"] = {
                "video": self._enrich_video_root_now,
                "music": self._enrich_audio_root_now,
                "photo": 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()
            console_shutdown_done = None
            if sys.platform == "win32":
                # SIGBREAK: CTRL_BREAK_EVENT, how platform.autostart_end() asks
                # a per-user-mode daemon to stop gracefully instead of only
                # ever taskkill /F.
                for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGBREAK):
                    signal.signal(sig, lambda *_: stop_event.set())
                # CTRL_CLOSE/LOGOFF/SHUTDOWN reach no Python signal at all --
                # see platform.install_console_close_handler for why this is
                # a separate mechanism rather than another signal.signal() line.
                from meshbay_node.platform import install_console_close_handler
                console_shutdown_done = install_console_close_handler(loop, stop_event)
            else:
                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: spawn(self._reload_config()))
            except (NotImplementedError, AttributeError):
                pass          # no SIGHUP on Windows; `reload` says so there
            await stop_event.wait()

        await self._shutdown()
        if console_shutdown_done is not None:
            # Releases the console-control handler's blocking wait (see
            # platform.install_console_close_handler) so it can return and
            # let Windows actually end the process for CTRL_CLOSE/LOGOFF/
            # SHUTDOWN, now that cleanup is genuinely done rather than just
            # started.
            console_shutdown_done.set()

    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 = await self._build_roots(group_cfg)
            except RootError as e:
                log.error("Group %r: %s — keeping the roots already loaded",
                          group_cfg.name, e)
                continue
            # `writable` and `removable` are in the comparison because an
            # operator editing node.toml by hand and reloading is a supported
            # way to change them, and a set compared on name and path alone
            # reports "nothing changed" for exactly that edit.
            if _root_shape(ctx["roots"]) == _root_shape(roots):
                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))
            # The new set is what the node serves from this moment, and the
            # scan of an added root is not waited for. Awaiting it here held
            # `_reload_lock` and the old set for as long as the scan ran —
            # hours for a large drive — so every file request under the new
            # root found no root to resolve against, and any op answering with
            # the live table (a writable/removable toggle) showed the directory
            # gone from the operator's settings.
            ctx["roots"] = roots
            await indexer.retarget(roots, wait=False)
            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 = await self._build_roots(group_cfg)
            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,
                on_root_ejected=self._eject_persister(group_cfg.id),
                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,
                "record_upload": indexer.record_upload,
                "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
                "debounce_secs": scan_settings["debounce_secs"],
                "visibility": group_cfg.visibility,
                "join_policy": group_cfg.join_policy,
                "enabled_apps": (
                    await self._roster.enabled_apps(group_cfg.id)
                    if self._roster else list(Roster.DEFAULT_APPS)),
                "transfer_limits": (
                    await self._roster.transfer_limits(group_cfg.id)
                    if self._roster else {}),
                **(await self._app_directories_ctx(group_cfg.id)),
                "chat_link_preview": (
                    await self._roster.chat_link_preview(group_cfg.id)
                    if self._roster else True),
                "chat_epoch": await self._ensure_chat_epoch(group_cfg.id),
                "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 (via `meshbay-node
                # status` or the desktop client, both of which query the control
                # API). Exiting would 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. Get it from `meshbay-node "
                            "status` and paste it in Settings > Link Node on %s "
                            "(the desktop client links it automatically). "
                            "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 _ensure_chat_epoch(self, group_id: str) -> int:
        """
        The group's current chat epoch, opening the first one if it has none.

        Chat is always encrypted, so a group with no epoch key is a group in
        which nobody can say anything. Attaching one is the node's job and
        happens here rather than on the first message: a failure at start-up is
        in the log the operator is already reading, and a failure on someone's
        first message is a chat that mysteriously refuses them.

        Never fatal. A group whose epoch cannot be opened keeps every other
        function — files, video, the index — and only its chat is unusable,
        which is strictly better than refusing to host the group at all.
        """
        if not self._bundle_store:
            return 0
        try:
            epoch = await self._bundle_store.latest_chat_epoch(group_id)
            if epoch:
                return epoch
            from meshbay_node import ops

            return (await ops.open_chat_epoch(self._state, group_id))["epoch"]
        except Exception as e:
            log.error("chat: no epoch key for group %s (%s) — chat is "
                      "unusable in this group until this is fixed",
                      group_id[:8], e)
            return 0

    async def _reap_partial_uploads(self, interval: float = 3600.0,
                                    first_delay: float = 60.0) -> None:
        """
        Delete `.part` files that no upload will ever finish.

        An upload interrupted for good leaves its partial file behind, and
        nothing else ever looks at it: `.part` is not an index entry, so it is
        invisible to every group member and to the operator's own file list. One
        abandoned film is a gigabyte of their disk, kept for ever.

        Two conditions, both required, and `uploads.orphaned_parts` is where
        they are stated and tested. What this adds is the walk and the deletion,
        and one rule of its own: it runs a minute after start rather than at
        once, so a client reconnecting to finish an upload that outlived a node
        restart is not raced by the janitor that would have deleted it — the age
        threshold makes that impossible in practice, and doing it anyway costs a
        minute.

        `interval` and `first_delay` are parameters so a test can drive this
        without waiting an hour.
        """
        await asyncio.sleep(first_delay)
        while True:
            try:
                self._reap_once()
            except Exception as exc:   # never let the janitor kill the node
                log.warning("Reaping partial uploads failed: %s", exc)
            await asyncio.sleep(interval)

    def _reap_once(self, now: float | None = None) -> int:
        """One pass over every group. Returns how many files were deleted."""
        groups = (self._webrtc._ctx.get("groups") or {}) if self._webrtc else {}
        when = time.time() if now is None else now
        deleted = 0
        for gid, ctx in groups.items():
            roots = ctx.get("roots")
            if roots is None:
                continue
            store = ctx.get("partial_uploads")
            live = store.live_paths() if store is not None else set()
            for path in uploads_mod.orphaned_parts(
                    uploads_mod.find_parts(roots.roots), live, when):
                try:
                    size = path.stat().st_size
                    path.unlink()
                except OSError as exc:
                    log.warning("Could not remove abandoned upload %s: %s",
                                path.name, exc)
                    continue
                deleted += 1
                log.info("Removed abandoned upload %s (%d bytes, group %s)",
                         path.name, size, gid[:8])
        return deleted

    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
        # Deliberately NOT sealed, unlike index_sync/index_delta (decision D3).
        # Counters only — never a path, never a filename, see IndexProgress in
        # indexer.py — pushed every couple of seconds for the whole length of a
        # scan. Sealing it would buy an attacker's rough estimate of a library's
        # size and cost a key derivation and a decrypt per push. If a field that
        # names anything is ever added here, that trade is void and this message
        # joins the other two.
        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 _build_roots(self, group_cfg) -> RootSet:
        """
        Build a group's RootSet from node.toml, with the ejected state restored.

        node.toml carries configuration (`writable`, `removable`); the roster
        carries the runtime answer to "is this drive ejected right now". They
        are merged here, in the one place every caller goes through, because a
        root that quietly comes back available across a restart is exactly the
        surprise unplug that eject exists to survive.
        """
        specs = [asdict(r) for r in group_cfg.roots]
        if self._roster:
            ejected = await self._roster.ejected_roots(group_cfg.id)
            if ejected:
                for spec in specs:
                    name = spec.get("name") or Path(spec.get("path", "")).name
                    if fold(name) in ejected:
                        spec["ejected"] = True
        return RootSet.build(specs)

    # Every application that keeps directories. This is the one list, and it
    # lives here because the daemon is what wires a group's context: `roster.py`,
    # `ops.py` and the rest must name no application at all — that is the
    # property the reference app exists to demonstrate
    # (`test_helloworld_proves_the_plugin_claim.py`).
    #
    # Not derived from `enabled_apps`: the context is read once at load, and an
    # application enabled later must not find its own setting missing.
    #
    # The handshake ack does **not** get a copy of this. It emits whatever
    # `<app>_directories` the context holds, so the two cannot drift — a copy
    # lived in `webrtc_server.py` until 2026-09-10 and had already lost
    # `helloworld`, which made the app that proves a new one needs no
    # special-casing the single app whose directories never reached a client.
    APP_DIR_KEYS = ("video", "music", "photo", "chat", "helloworld")

    async def _app_directories_ctx(self, group_id: str) -> dict:
        """
        Each app's configured directories, plus the second name an app is
        also published under where something reads one (`chat_directory`).

        Derived here rather than stored, so the two can never disagree.
        """
        dirs = {}
        for app in self.APP_DIR_KEYS:
            dirs[f"{app}_directories"] = (
                await self._roster.app_directories(group_id, app)
                if self._roster else [])
        aliases = {}
        for app in self.APP_DIR_KEYS:
            alias = Roster.ctx_alias(app, dirs[f"{app}_directories"])
            if alias:
                aliases[alias[0]] = alias[1]
        return {**dirs, **aliases}

    def _eject_persister(self, group_id: str):
        """`on_root_ejected` bound to one group, for that group's indexer."""
        async def persist(root_name: str, ejected: bool) -> None:
            if self._roster:
                await self._roster.set_root_ejected(
                    group_id, root_name, ejected,
                    set_by=self._state.get("node_user_id", ""))
        return persist

    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)
            spawn(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)

        # A root that was ejected and plugged back in, or that fell off and
        # re-mounted, has had its entries thrown away and rebuilt from disk
        # (`indexer._drop_root_entries`). The rebuilt entry has the same
        # content-hash id and none of the enrichment fields, so the diff above
        # reports neither an addition nor a deletion — and `_enriched_attempted`
        # still says "done" for a file whose album and cover no longer exist.
        # Found live: a Music library came back with its files and without its
        # albums, and stayed that way, because only a restart (which starts
        # with no snapshot, making every entry an addition) could clear either
        # gate. Treated here as what it is — those entries are new again.
        rebuilt_ids = indexer.drain_rescanned_ids()
        if rebuilt_ids:
            rebuilt = [e for e in idx.entries if e.id in rebuilt_ids]
            for entry in rebuilt:
                self._enriched_attempted.discard((group_id, entry.id))
            seen = {e.id for e in new_entries}
            new_entries = new_entries + [e for e in rebuilt if e.id not in seen]
        spawn(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).
        spawn(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).
        spawn(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:
            spawn(
                self._reenrich_renamed_video_entries(indexer, delta.updates, previous))
            spawn(
                self._reenrich_renamed_audio_entries(indexer, delta.updates, previous))
            spawn(
                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:
                    spawn(self._media_cache.prune_file(file_id))

        # 11.5 — Push to connected WebRTC peers in this group
        if self._webrtc:
            peers = [s for s in list(self._webrtc._sessions.values())
                     if s._group_id == group_id]
            # Both messages are sealed under a GEK-derived subkey, so building one
            # needs a key. A group without one has no peers to push to either — the
            # node refuses every handshake while the GEK is None (NS8) — so this is
            # "nobody is listening", not a case to send in clear for.
            if peers and idx.gek:
                msg = (index_delta_message(idx, delta, indexer.roots)
                       if delta is not None
                       else index_sync_message(idx, indexer.roots))
                pushed = 0
                for session in peers:
                    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}"
                spawn(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 when it changes)
        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_dirs = await self._roster.app_directories(indexer.group_id, "video")
        if not video_dirs:
            return
        for entry in entries:
            if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted:
                continue
            if not _under_any_directory(entry.path, video_dirs):
                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.

        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_dirs = await self._roster.app_directories(indexer.group_id, "music")
        if not audio_dirs:
            return
        # Resolved once per directory, not once per file: a library is
        # thousands of entries and this is a filesystem call each time.
        boundaries = {d: indexer.roots.resolve(d, require_available=False)
                      for d in audio_dirs}
        for entry in entries:
            if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted:
                continue
            owner = _owning_directory(entry.path, audio_dirs)
            if owner is None:
                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)

            # The boundary is *the configured directory this file is under*,
            # not the shared root it lives in — so the ancestor walk
            # (enrich_audio._artist_album_from_ancestors) treats a flat
            # top-level folder right under the configured Music directory as
            # ambiguous (artist-or-release, musicbay.md §2.1), rather than one
            # level too shallow when that directory is itself a subfolder.
            # With several configured, each file is measured against its own:
            # a single shared boundary would be wrong for all but one of them.
            self._audio_enricher.spawn(entry, file_path, on_done,
                                       boundaries.get(owner))

    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. 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_dirs = await self._roster.app_directories(indexer.group_id, "photo")
        if not photo_dirs:
            return
        for entry in entries:
            if entry.type != "image" or (indexer.group_id, entry.id) in self._enriched_attempted:
                continue
            if not _under_any_directory(entry.path, photo_dirs):
                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 change. 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:
                spawn(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 control API's per-run
    session token (11.5.3), the same gate the desktop client's Node page passes.
    """
    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(encoding="utf-8").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.

    Reloads the user manager's view of unit files first. The package
    installers (deb postinst, rpm %post) run as root and can only reload the
    *system* manager — a different process from any signed-in user's *user*
    manager, which is the one that actually owns this unit — so a package
    upgrade leaves that manager still holding the old unit file and prints a
    warning naming the exact fix. Doing it here runs it under the right
    privilege, the user's own, right before the command that would otherwise
    act on a stale definition. Best-effort and unchecked: a reload the
    manager did not need must never block what the operator actually asked
    for, and a genuine problem still surfaces from the verb below.
    """
    import subprocess

    subprocess.run(["systemctl", "--user", "daemon-reload"],
                   capture_output=True, text=True)

    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

    from meshbay_node.platform import (configure_event_loop, force_utf8_stdio,
                                        load_node_env)
    force_utf8_stdio()
    configure_event_loop()
    # Before anything reads the environment. On Linux systemd has usually loaded
    # the same file already via EnvironmentFile=; this is what makes a Windows
    # run (Startup-folder .vbs, no systemd) and a bare `meshbay-node` behave the
    # same. Already-set variables are left alone, so it cannot undo either.
    load_node_env(config_dir())

    parser = argparse.ArgumentParser(description="MeshBay Node daemon")
    parser.add_argument("command", nargs="?",
                        choices=["init", "reset", "status", "gek-init",
                                 "gek", "operator", "member", "group", "root",
                                 "file", "video", "chat", "denylist", "stun",
                                 "transfers",
                                 "reload",
                                 "restart-daemon", "autostart", "service",
                                 "calibrate-argon2"],
                        help="init: provision config + keystore | reset: erase all "
                             "node state | status: node state and keys "
                             "| operator pair: pair a "
                             "browser with this node | member list|invite|revoke|unpin "
                             "| group list|add|remove "
                             "| root list|add|remove|set|eject|plug "
                             "| gek init|rotate | file list|rm "
                             "| video rematch: re-resolve TMDB matches for a group's "
                             "videos "
                             "| chat status|rotate|encrypt-history|prune "
                             "| denylist show|clear "
                             "| stun list|add|remove|reset "
                             "| transfers show|set|per-member: live transfer "
                             "slots, the node-wide caps, and how many one "
                             "member may run at once in a group "
                             "| reload: re-read node.toml (hot; systemd or the "
                             "loopback API) | restart-daemon: restart the node "
                             "(systemd unit, the Windows autostart launcher, or the "
                             "service task, whichever applies) "
                             "| autostart install|remove|start|stop|status "
                             "(Windows: run meshbay-node at each sign-in, no admin) "
                             "| service install|remove|start|stop|status "
                             "(Windows: run at boot, before sign-in, needs admin "
                             "once to install) "
                             "| calibrate-argon2: benchmark")
    parser.add_argument("subcommand", nargs="?",
                        help="'pair' for operator; list|invite|revoke|unpin for "
                             "member; list|add|remove for group; "
                             "list|add|remove|set|eject|plug for root; "
                             "init|rotate for gek; "
                             "list|rm for file; rematch for video; show|clear for "
                             "denylist; list|add|remove|reset for stun; "
                             "show|set|per-member for transfers; "
                             "install|remove|start|stop|status for autostart and "
                             "for service")
    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; download cap for transfers set")
    parser.add_argument("value", nargs="?",
                        help="the second value where a verb takes two: the "
                             "upload cap for transfers set")
    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("--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("--writable", action="store_true", default=None,
                        dest="writable",
                        help="root accepts member uploads (root add/set)")
    parser.add_argument("--no-writable", action="store_false",
                        dest="writable",
                        help="root is read-only (root add/set, group add)")
    parser.add_argument("--removable", action="store_true", default=None,
                        dest="removable",
                        help="mark root as removable (root set/add)")
    parser.add_argument("--no-removable", action="store_false",
                        dest="removable",
                        help="mark root as not removable (root set)")
    parser.add_argument("--name", default=None,
                        help="root name (root add; defaults to directory basename)")
    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", "gek-init", "gek", "operator",
                             "member", "group", "root", "file", "video", "chat",
                             "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
        cfg_dir = cfg_path.parent
        cfg_dir.mkdir(parents=True, exist_ok=True)

        from meshbay_node.platform import install_node_env
        env_written = install_node_env(cfg_dir)
        if env_written:
            print(f"Wrote {env_written} (packaged defaults).")

        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(encoding="utf-8")
            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, encoding="utf-8", newline="\n")
                print(f"Config updated: username={username}, hub={hub_url}")
            else:
                print(f"Config already exists: {cfg_path}")
        else:
            unlock_file = cfg_dir / "unlock.key"
            toml_lines = [
                "[hub]",
                f'url      = "{hub_url}"',
                f'username = "{username}"',
                "",
                "[node]",
                "quic_enabled = false  # QUIC direct path; no client uses it yet",
                "quic_port = 19010",
                "ui_port   = 18000",
                "",
                "[keystore]",
                # Forward slashes: a Windows path in a TOML basic string is a
                # parse error (`\U`, `\a`, ... are escape sequences).
                f'unlock_file = "{unlock_file.as_posix()}"',
                "",
            ]
            cfg_path.write_text("\n".join(toml_lines) + "\n", encoding="utf-8", newline="\n")
            chmod_private(cfg_path)
            print(f"Config written to {cfg_path}")

        unlock_file = cfg_dir / "unlock.key"
        if not unlock_file.exists():
            import secrets
            key = secrets.token_urlsafe(32)
            unlock_file.write_text(key + "\n", encoding="utf-8", newline="\n")
            chmod_private(unlock_file)
            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")
        if sys.platform == "win32":
            print("  2. meshbay-node autostart install   (run at each sign-in)")
            print("     — or just: meshbay-node          (start it now, this session)")
        else:
            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_ = config_dir()
        data_dir_ = data_dir()
        state_dir_ = state_dir()

        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(encoding="utf-8").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).")

        if sys.platform == "win32":
            from meshbay_node.platform import autostart_remove, service_remove
            autostart_remove()
            service_remove()   # no-op, silently, if not elevated or not installed
        else:
            _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(encoding="utf-8").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)}")

            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
                    flags = []
                    if getattr(r, 'writable', False) or getattr(r, 'upload', False):
                        flags.append("rw")
                    else:
                        flags.append("ro")
                    if getattr(r, 'removable', False):
                        flags.append("removable")
                    flag_str = f"  ({', '.join(flags)})" if flags else ""
                    live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
                    print(f"             {label} → {r.path}{flag_str}{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

        # `member upload` is gone: whether uploads are accepted is `writable`
        # on the root they would land in, not a per-group switch. Named
        # explicitly rather than left to the usage line below, which offered a
        # username for a verb that no longer takes one — an operator following
        # it would have got "unknown subcommand" and no idea what replaced it.
        if sub == "upload":
            print("`member upload` is gone. Uploads are decided per directory "
                  "now:")
            print()
            print("  meshbay-node root list                    "
                  "# which are read-write")
            print("  meshbay-node root set <name> --writable   "
                  "# accept uploads there")
            print("  meshbay-node root set <name> --no-writable  # stop them")
            print()
            print("A group whose directories are all read-only accepts no "
                  "uploads at all,")
            print("which is what turning the old switch off meant.")
            sys.exit(1)

        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":
        if sys.platform == "win32":
            # No systemd, no SIGHUP: the daemon exposes a hot reload on its
            # own loopback API (the same one ops.reload_config drives).
            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
            _daemon_api(cfg, "/api/reload", method="POST")
            print("sent reload to the running node")
            return
        _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":
        if sys.platform == "win32":
            from meshbay_node.platform import (
                autostart_end, autostart_run, service_end, service_run, service_status,
            )
            if service_status()["installed"]:
                service_end()
                service_run()
                print("restarted the node (service task)")
                return
            autostart_end()   # kill whatever is running now
            try:
                autostart_run()
            except RuntimeError as e:
                print(f"Could not restart: {e}. Stop the daemon (Ctrl+C) and "
                      "relaunch it from where meshbay-node is on PATH.")
                sys.exit(1)
            print("restarted the node")
            return
        _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 == "autostart":
        from meshbay_node import platform as _plat
        if sys.platform != "win32":
            print("autostart is Windows-only — elsewhere use "
                  "'systemctl --user enable --now meshbay-node'.")
            sys.exit(1)
        sub = args.subcommand or "status"
        if sub == "install":
            _plat.autostart_install()
            print("Installed the Startup launcher — meshbay-node starts at "
                  "each sign-in (no window, no admin).")
            print("Start it now with: meshbay-node autostart start")
        elif sub == "remove":
            _plat.autostart_remove()
            print("Removed the Startup launcher.")
        elif sub == "start":
            try:
                _plat.autostart_run()
            except RuntimeError as e:
                print(f"Could not start: {e}")
                sys.exit(1)
            print("started")
        elif sub == "stop":
            _plat.autostart_end()
            print("stopped")
        elif sub == "status":
            st = _plat.autostart_status()
            if st["installed"]:
                print("autostart    installed — runs meshbay-node at sign-in")
            else:
                print("autostart    not installed — meshbay-node autostart install")
        else:
            print("autostart: install | remove | start | stop | status")
            sys.exit(1)
        return

    if args.command == "service":
        from meshbay_node import platform as _plat
        if sys.platform != "win32":
            print("service mode is Windows-only — elsewhere use "
                  "'systemctl --user enable --now meshbay-node'.")
            sys.exit(1)
        sub = args.subcommand or "status"
        if sub == "install":
            try:
                _plat.service_install()
            except RuntimeError as e:
                print(f"Could not install: {e}")
                if "denied" in str(e).lower():
                    print("Run this from an elevated (Administrator) prompt.")
                sys.exit(1)
            print(f"Registered the {_plat.TASK_NAME!r} scheduled task — it starts "
                  "meshbay-node at boot, as this user, whether or not you have "
                  "signed in yet (no password stored).")
            print("Start it now with: meshbay-node service start")
        elif sub == "remove":
            _plat.service_remove()
            print(f"Removed the {_plat.TASK_NAME!r} scheduled task.")
        elif sub == "start":
            _plat.service_run()
            print("started")
        elif sub == "stop":
            _plat.service_end()
            print("stopped")
        elif sub == "status":
            st = _plat.service_status()
            if st["installed"]:
                print(f"service       installed — {st['state'] or 'unknown state'}")
            else:
                print("service       not installed — meshbay-node service install "
                      "(needs an elevated prompt)")
        else:
            print("service: install | remove | start | stop | status")
            sys.exit(1)
        return

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

        if sub == "status":
            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat")
            print(f"encryption   always on (MNP {MNP_VERSION})")
            print(f"epoch        {out.get('epoch', 0)}")
            print(f"messages     {out.get('encrypted_messages', 0)} encrypted, "
                  f"{out.get('plaintext_messages', 0)} in the clear")
            if out.get("plaintext_messages"):
                print("\nThose messages were written before this node spoke MNP "
                      "2.0 and are\nstill readable off this disk. "
                      "`chat encrypt-history` converts them.")
            return

        if sub == "rotate":
            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/epoch",
                              method="POST")
            print(f"chat epoch {out['epoch']} opened")
            print("Everyone still in the group keeps reading the history; "
                  "whoever left\ncannot read what is written from now on.")
            return

        if sub == "encrypt-history":
            if not args.yes:
                print("This rewrites the only copy of this group's older "
                      "messages.")
                print("A backup of chat.db is taken first, beside it.")
                if input("re-encrypt now? [y/N] ").strip().lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            out = _daemon_api(cfg, f"/api/groups/{group_id}/chat/encrypt-history",
                              method="POST", timeout=300)
            print(f"re-encrypted {out['converted']} message(s) under epoch "
                  f"{out['epoch']}")
            print(f"backup       {out['backup']}")
            return

        if sub == "prune":
            days = int(args.target or 0)
            if days < 1:
                print("usage: meshbay-node chat prune <days> [--group G]")
                sys.exit(1)
            out = _daemon_api(
                cfg, f"/api/groups/{group_id}/chat/prune?max_age_days={days}",
                method="POST")
            print(f"removed {out['removed']} message(s) older than {days} day(s)")
            return

        print("usage: meshbay-node chat "
              "status|rotate|encrypt-history|prune [--group G]")
        sys.exit(1)

    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 == "transfers":
        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        sub = args.subcommand or "show"

        if sub == "show":
            out = _daemon_api(cfg, "/api/transfers")
            for kind, pool in out.get("pools", {}).items():
                print(f"  {kind:<9} {pool['in_use']}/{pool['cap']} in use, "
                      f"{pool['queued']} queued   (node-wide)")
            # Per group, because that is the cap that decides how many one
            # person runs at once — and it is not the node-wide number. An
            # operator raising `transfers set 8 8` and still seeing two at a
            # time is looking at this line, which used to print the node's
            # default and say nothing about where it came from.
            groups = out.get("groups") or []
            if groups:
                print("\n  per member, per group "
                      "(meshbay-node transfers per-member <dl> <ul> --group X):")
                for g in groups:
                    how = "set" if g["set"] else "default"
                    print(f"    {g['name']:<20} {g['download']} download(s), "
                          f"{g['upload']} upload(s)   [{how}]")
            leases = out.get("leases", [])
            if not leases:
                print("\n  nothing transferring")
                return
            print(f"\n  {'transfer':<14}{'kind':<10}{'state':<9}"
                  f"{'user':<12}{'bytes':>12}")
            for x in leases:
                where = f"  (#{x['ahead'] + 1} in queue)" if x["state"] == "queued" else ""
                print(f"  {x['tr']:<14}{x['kind']:<10}{x['state']:<9}"
                      f"{x['user_id'][:10]:<12}{x['bytes']:>12}{where}")
            return

        if sub == "set":
            # `transfers set 4 2` — downloads, then uploads. Node-wide; the
            # per-member cap is a group's setting and is signed, so it is not
            # settable from here (see `ops.set_transfer_limits`).
            values = [v for v in (args.target, args.value) if v]
            if len(values) != 2:
                print("usage: meshbay-node transfers set <downloads> <uploads>")
                sys.exit(1)
            try:
                downloads, uploads = int(values[0]), int(values[1])
            except ValueError:
                print("error: both values must be whole numbers")
                sys.exit(1)
            if downloads < 1 or uploads < 1:
                print("error: a cap below 1 is not 'unlimited'; it would stop "
                      "every transfer. Revoke the member instead.")
                sys.exit(1)
            out = _daemon_api(cfg, "/api/node-settings", method="PUT",
                              body={"max_concurrent_downloads": downloads,
                                    "max_concurrent_uploads": uploads})
            print(f"downloads: {downloads}, uploads: {uploads} "
                  f"(applied now, and kept in node.toml)")
            return

        if sub == "per-member":
            # How many transfers ONE member may run at once in this group. Not
            # the same knob as `set`, which is the machine's total — and the
            # reason "I set 8 8 and still only get two" is the commonest
            # confusion here: per-member is checked first, by design.
            values = [v for v in (args.target, args.value) if v]
            if len(values) != 2:
                print("usage: meshbay-node transfers per-member <downloads> "
                      "<uploads> [--group NAME]")
                sys.exit(1)
            try:
                downloads, uploads = int(values[0]), int(values[1])
            except ValueError:
                print("error: both values must be whole numbers")
                sys.exit(1)
            if downloads < 1 or uploads < 1:
                print("error: a cap below 1 is not 'unlimited'; it would stop "
                      "every transfer for that member. Revoke them instead.")
                sys.exit(1)
            group_id = _resolve_group(cfg, args.group)
            out = _daemon_api(cfg, f"/api/groups/{group_id}/transfer-limits",
                              method="PUT",
                              body={"downloads": downloads, "uploads": uploads})
            got = out.get("limits", {})
            started = out.get("started") or []
            print(f"each member of this group may now run "
                  f"{got.get('download')} download(s) and "
                  f"{got.get('upload')} upload(s) at once")
            if started:
                print(f"{len(started)} waiting transfer(s) started at once")
            return

        print("usage: meshbay-node transfers show|set <downloads> <uploads>|"
              "per-member <downloads> <uploads> [--group NAME]")
        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("writable"):
                        flags.append("rw")
                    else:
                        flags.append("ro")
                    if r.get("removable"):
                        flags.append("removable")
                    if r.get("ejected"):
                        flags.append("ejected")
                    flag_str = f" ({', '.join(flags)})" if flags else ""
                    live = "" if r.get("available", True) else "  [UNAVAILABLE]"
                    print(f"      root {r['name']}{flag_str}{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> "
                  "[--no-writable]")
            print()
            print("The group must already exist on the hub and be yours. This")
            print("only tells the node to host it, and picks its first")
            print("directory, which accepts uploads unless --no-writable.")
            print("Add more with:  meshbay-node root add <path> [--writable]")
            sys.exit(1)

        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
        # Writable unless the operator says otherwise: a brand-new group that
        # cannot receive a single file until its owner finds a second command
        # is not a working group. Every root added *later* is read-only by
        # default, which is the opposite rule and the right one there.
        writable = args.writable is not False
        body = {"name": args.target, "shared_dir": args.dir,
                "writable": writable}
        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']}"
              f"  ({'read-write' if writable else 'read-only'})")
        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 == "root":
        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, "/api/groups")
            group = next((g for g in out.get("groups", [])
                          if g["id"] == group_id), None)
            if not group:
                print(f"group {group_id[:8]} not hosted on this node")
                sys.exit(1)
            roots = group.get("roots", [])
            if not roots:
                print("no roots configured")
                print(f"add one:  meshbay-node root add /path/to/dir --group {group_id}")
                return
            for r in roots:
                flags = []
                if r.get("writable"):
                    flags.append("rw")
                else:
                    flags.append("ro")
                if r.get("removable"):
                    flags.append("removable")
                if r.get("ejected"):
                    flags.append("EJECTED")
                avail = "available" if r.get("available", True) else "UNAVAILABLE"
                flags.append(avail)
                print(f"  {r['name']:<20} {', '.join(flags)}")
                print(f"    {r.get('path', '?')}")
            return

        if sub == "add":
            path = args.target
            if not path:
                print("usage: meshbay-node root add <path> [--name NAME] "
                      "[--writable] [--removable] [--group NAME]")
                sys.exit(1)
            body = {
                "path": path,
                "name": args.name or Path(path).name,
                "writable": args.writable if args.writable is not None else True,
                "removable": bool(args.removable),
            }
            _daemon_api(cfg, f"/api/groups/{group_id}/roots",
                        method="POST", body=body)
            w = "rw" if body["writable"] else "ro"
            rm = ", removable" if body["removable"] else ""
            print(f"added root {body['name']} → {path} ({w}{rm})")
            print("reload the daemon to start indexing:")
            print("  meshbay-node reload")
            return

        if sub == "remove":
            name = args.target
            if not name:
                print("usage: meshbay-node root remove <name> [--group NAME]")
                sys.exit(1)
            if not args.yes:
                print(f"Remove root '{name}' from group {group_id[:8]}?")
                print("Files on disk are untouched; only the node config changes.")
                if input("remove? [y/N] ").strip().lower() not in ("y", "yes"):
                    print("cancelled")
                    return
            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}",
                        method="DELETE")
            print(f"removed root {name}")
            print("reload the daemon to apply:")
            print("  meshbay-node reload")
            return

        if sub == "set":
            name = args.target
            if not name:
                print("usage: meshbay-node root set <name> "
                      "[--writable|--no-writable] "
                      "[--removable|--no-removable] [--group NAME]")
                sys.exit(1)
            body = {}
            if args.writable is not None:
                body["writable"] = args.writable
            if args.removable is not None:
                body["removable"] = args.removable
            if not body:
                print("nothing to change — pass --writable/--no-writable "
                      "or --removable/--no-removable")
                sys.exit(1)
            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}",
                        method="PATCH", body=body)
            changes = ", ".join(f"{k}={v}" for k, v in body.items())
            print(f"updated root {name}: {changes}")
            return

        if sub == "eject":
            name = args.target
            if not name:
                print("usage: meshbay-node root eject <name> [--group NAME]")
                sys.exit(1)
            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/eject",
                        method="PUT")
            print(f"ejected root {name} — files are hidden until plugged back")
            return

        if sub == "plug":
            name = args.target
            if not name:
                print("usage: meshbay-node root plug <name> [--group NAME]")
                sys.exit(1)
            _daemon_api(cfg, f"/api/groups/{group_id}/roots/{name}/plug",
                        method="PUT")
            print(f"plugged root {name} — files are visible again")
            return

        print("usage: meshbay-node root list|add|remove|set|eject|plug [name] "
              "[--group NAME]")
        sys.exit(1)

    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

    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)

    from meshbay_node.platform import check_media_tools
    try:
        check_media_tools(cfg.node.ffmpeg_path, cfg.node.ffprobe_path)
    except RuntimeError as e:
        print(f"Error: {e}")
        sys.exit(1)

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


if __name__ == "__main__":
    main()