feat(node): add audio_root, gate Music enrichment on it like video_root

musicbay.md's original call — Music needs no root, tag reads are cheap so
just cover the whole shared tree — didn't hold up against a real messy
library: everything under every shared folder got mixed together with no
way to scope Music down to an actual music collection. This adds an
audio_root setting, symmetric to video_root in every respect: signed
operator op (audio_root/audio_root_ack, MNP bumped to 0.10), validated
against a real directory in the group's own roots before a signature is
even asked for, gates tag/cover enrichment exactly like video_root gates
ffprobe/TMDB (nothing runs until it's set, only files under it once it
is), and a set/change fires a one-off sweep of whatever the folder already
contains.

The old trigger — sweep everything the instant "music" joins enabled_apps
— is gone along with the root-less design it belonged to; setting
audio_root is now the trigger, mirroring set_video_root's
enrich_video_root_fn exactly.

Test coverage mirrors the video_root suite: policy (refuse before a
signature round trip, accept/store correctly) and the enrichment gate
itself (nothing without a root, only files under it, sweep on set).

4 files changed, 128 insertions, 40 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index ae4e1f0..0015f1a 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -73,6 +73,12 @@ def _under_video_root(path: str, video_root: str) -> bool:
     return path == video_root or path.startswith(video_root + "/")
 
 
+def _under_audio_root(path: str, audio_root: str) -> bool:
+    """Mirrors music-app.js's underAudioRoot — same shape as _under_video_root."""
+    path = path or ""
+    return path == audio_root or path.startswith(audio_root + "/")
+
+
 # ── Argon2id calibration ──────────────────────────────────────────────────────
 
 def calibrate_argon2(target_ms: int = 500) -> None:
@@ -340,6 +346,9 @@ class NodeDaemon:
                     # group — "" means the whole group index.
                     "video_root": await self._roster.video_root(
                         group_cfg.id) if self._roster else "",
+                    # Same shape, Music app's own entry point.
+                    "audio_root": await self._roster.audio_root(
+                        group_cfg.id) if self._roster else "",
                     # 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"
@@ -541,7 +550,7 @@ class NodeDaemon:
             self._state["hub"] = hub
             self._state["reload_fn"] = self._reload_config
             self._state["enrich_video_root_fn"] = self._enrich_video_root_now
-            self._state["enrich_music_now_fn"] = self._enrich_music_now
+            self._state["enrich_audio_root_fn"] = self._enrich_audio_root_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
@@ -750,6 +759,9 @@ class NodeDaemon:
                 "video_root": (
                     await self._roster.video_root(group_cfg.id)
                     if self._roster else ""),
+                "audio_root": (
+                    await self._roster.audio_root(group_cfg.id)
+                    if self._roster else ""),
                 "tmdb_enabled": (
                     await self._roster.tmdb_enabled(group_cfg.id)
                     if self._roster else True),
@@ -988,11 +1000,9 @@ class NodeDaemon:
         # INDEX_DELTA update (_on_enriched below).
         new_entries = delta.additions if delta is not None else list(idx.entries)
         asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries))
-        # Music app (docs/musicbay.md §6): same shape, gated on the group's
-        # enabled_apps rather than a root (Music has no video_root analogue —
-        # see musicbay.md §2.1's note on why that scoping wasn't carried
-        # over). Tag/cover extraction is local and cheap either way; the gate
-        # exists to not do it at all for a group that never turned Music on.
+        # Music app (docs/musicbay.md §6): same shape, gated on audio_root
+        # exactly like video_root above (added later — musicbay.md's
+        # original "no root, whole shared tree" call didn't hold up).
         asyncio.ensure_future(self._enrich_new_audio_entries(indexer, new_entries))
 
         # A rename/move changes the very filename (or season folder) that
@@ -1147,49 +1157,49 @@ class NodeDaemon:
     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, gated on the
-        group having "music" in its enabled_apps — there is no video_root
-        analogue for Music (musicbay.md §2.1 deliberately didn't add one:
-        tag reads are free/local, unlike ffprobe+ffmpeg thumbnailing, so the
-        cost this gate protects against is smaller, and most personal MP3
-        libraries want their whole shared tree available rather than one
-        scoped subfolder). `_enriched_attempted` is shared with the video
-        path — content-addressed ids never collide across the two.
+        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
-        enabled_apps = await self._roster.enabled_apps(indexer.group_id)
-        if "music" not in enabled_apps:
+        audio_root = await self._roster.audio_root(indexer.group_id)
+        if not audio_root:
             return
+        root_boundary = indexer.roots.resolve(audio_root, require_available=False)
         for entry in entries:
             if entry.type != "audio" or entry.id in self._enriched_attempted:
                 continue
+            if not _under_audio_root(entry.path, audio_root):
+                continue
             file_path = entry_abs_path(indexer.roots, entry)
             if not file_path or not file_path.exists():
                 continue
             self._enriched_attempted.add(entry.id)
-            # The entry's own named root, so the ancestor walk
-            # (enrich_audio._artist_album_from_ancestors) can tell "a
-            # top-level folder under this root" from "one level deeper" —
-            # without this boundary it used to read the root's own
-            # directory name as an artist for every flat top-level folder.
-            split = indexer.roots.split(entry.path)
-            root_path = split[0].path if split else None
 
             async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
                 await self._on_enriched(_indexer, file_id, fields)
 
-            self._audio_enricher.spawn(entry, file_path, on_done, root_path)
+            # `root_boundary` — audio_root itself, not the shared root it
+            # lives under — so the ancestor walk
+            # (enrich_audio._artist_album_from_ancestors) treats a flat
+            # top-level folder right under the *configured* Music root as
+            # ambiguous (artist-or-release, §2.1), not one level too shallow
+            # if audio_root is itself a subfolder of a larger shared root.
+            self._audio_enricher.spawn(entry, file_path, on_done, root_boundary)
 
-    async def _enrich_music_now(self, group_id: str) -> None:
+    async def _enrich_audio_root_now(self, group_id: str) -> None:
         """
-        Music app: sweep a group's existing index right after "music" is
-        added to its enabled_apps (ops.set_enabled_apps) — the ordinary path
-        above only ever looks at entries new since the last broadcast, so a
-        library that was already sitting there before Music was turned on
-        would otherwise never get enriched. Mirrors
-        `_enrich_video_root_now`, triggered from a different setting because
-        Music has no root of its own to key off.
+        Music app: sweep a group's existing index right after its
+        audio_root is set or changed (ops.set_audio_root). Mirrors
+        `_enrich_video_root_now` exactly — the ordinary path above only
+        ever looks at entries new since the last broadcast, so a folder
+        that already had files in it before it became the audio_root would
+        otherwise never get enriched at all.
         """
         indexer = self._state.get("indexers", {}).get(group_id)
         if not indexer:
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index d439ec3..1da6928 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -728,14 +728,6 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict:
                                   set_by=state.get("node_user_id", ""))
     ctx["enabled_apps"] = apps
     log.info("Enabled apps for group %s: %s", group_id[:8], ",".join(sorted(apps)))
-    if "music" in apps:
-        # Music has no root of its own to key a sweep off (unlike
-        # set_video_root's enrich_video_root_fn) — turning the app on at all
-        # is the trigger, mirroring that same "sweep what's already there"
-        # need (docs/musicbay.md §2.1, daemon._enrich_music_now).
-        enrich_fn = state.get("enrich_music_now_fn")
-        if enrich_fn:
-            asyncio.ensure_future(enrich_fn(group_id))
     return {"apps": apps, "group_id": group_id}
 
 
@@ -850,6 +842,27 @@ async def set_video_root(state: dict, group_id: str, path: str) -> dict:
     return {"path": path, "group_id": group_id}
 
 
+async def set_audio_root(state: dict, group_id: str, path: str) -> dict:
+    """
+    Same shape as set_video_root above — the Music app's own entry point,
+    added later (docs/musicbay.md's original "no root, works over the
+    whole shared tree" simplification didn't hold up against a real messy
+    library). `path=""` clears it — the Music tab then asks for one to be
+    chosen before anything (including tag/cover enrichment) runs, rather
+    than defaulting to the whole shared index.
+    """
+    roster = _roster(state)
+    ctx = _group_ctx(state, group_id)
+    await roster.set_audio_root(group_id, path, set_by=state.get("node_user_id", ""))
+    ctx["audio_root"] = path
+    log.info("Music root for group %s: %r", group_id[:8], path)
+    if path:
+        enrich_fn = state.get("enrich_audio_root_fn")
+        if enrich_fn:
+            asyncio.ensure_future(enrich_fn(group_id))
+    return {"path": path, "group_id": group_id}
+
+
 # ── Scan settings ────────────────────────────────────────────────────────────
 
 async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float,
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index 24424ab..91e18cd 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -649,6 +649,21 @@ class Roster:
         await self.set_setting(group_id, self.SETTING_VIDEO_ROOT, path or "", set_by)
         return path or ""
 
+    # Same shape as SETTING_VIDEO_ROOT — the Music app's own entry point,
+    # added later (docs/musicbay.md's original "no root, works over the
+    # whole shared tree" simplification turned out not to hold up against a
+    # real messy library: the operator asked for the same scoping Videos
+    # already had). Empty/unset means Music shows nothing yet, exactly like
+    # an unset video_root — see daemon.py's enrichment gate.
+    SETTING_AUDIO_ROOT = "audio_root"
+
+    async def audio_root(self, group_id: str) -> str:
+        return await self.get_setting(group_id, self.SETTING_AUDIO_ROOT, "") or ""
+
+    async def set_audio_root(self, group_id: str, path: str, set_by: str = "") -> str:
+        await self.set_setting(group_id, self.SETTING_AUDIO_ROOT, path or "", set_by)
+        return path or ""
+
     # Whether TMDB lookups run for this group at all — per-group, unlike the
     # token/language above: one node process can share a real media library
     # group and several test/demo groups, and outbound TMDB traffic (and API
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 1629801..a5cccaa 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -72,6 +72,7 @@ from meshbay_common.adminop import (
     OP_TMDB_OVERRIDE,
     OP_MUSICBRAINZ_CONFIG,
     OP_MUSICBRAINZ_ENABLED,
+    OP_AUDIO_ROOT,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
     OP_GROUP_ATTACH,
@@ -398,6 +399,8 @@ class WebRTCPeerSession:
                 self._do_tmdb_enabled(msg)
             elif mtype == MNP.VIDEO_ROOT:
                 self._do_video_root(msg)
+            elif mtype == MNP.AUDIO_ROOT:
+                self._do_audio_root(msg)
             elif mtype == MNP.MEDIA_META_REQ:
                 self._spawn(self._do_media_meta_request(msg))
             elif mtype == MNP.SEASON_META_REQ:
@@ -685,6 +688,10 @@ class WebRTCPeerSession:
             "musicbrainz_enabled": bool(self._group_ctx().get("musicbrainz_enabled", True)),
             "musicbrainz_contact_configured": bool(
                 self._ctx.get("daemon_state", {}).get("musicbrainz_contact_configured", False)),
+            # Which folder the Music app treats as its entry point for this
+            # group — same shape as video_root above, "" means unset (the
+            # Music tab shows nothing yet).
+            "audio_root": self._group_ctx().get("audio_root") or "",
             # So a client that connects mid-scan shows the indexing state
             # immediately, instead of waiting for the next periodic
             # INDEX_PROGRESS push. Never a path or filename — see
@@ -1852,6 +1859,46 @@ class WebRTCPeerSession:
             except Exception:
                 pass
 
+    def _do_audio_root(self, msg: dict) -> None:
+        """Same shape as _do_video_root above — the Music app's own entry point."""
+        path = msg.get("path")
+        if not isinstance(path, str):
+            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
+            return
+        path = path.strip("/")
+        if path:
+            ctx = self._group_ctx()
+            resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None
+            if not resolved or not resolved.is_dir():
+                self._send({"type": "error", "detail": "Not a directory in this group"})
+                return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_AUDIO_ROOT, path)
+
+    async def _admin_exec_audio_root(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        path = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"audio_root:{path}")
+            return
+        try:
+            await self._run_op(ops.set_audio_root, self._group_id or "", path)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("audio_root", path)
+
+        notice = {"type": MNP.AUDIO_ROOT_ACK, "v": MNP_VERSION, "path": path}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     def _do_musicbrainz_config(self, msg: dict) -> None:
         """
         Set (or clear) the node-wide MusicBrainz User-Agent contact string
@@ -3551,6 +3598,9 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_MUSICBRAINZ_ENABLED:
             self._spawn(
                 self._admin_exec_musicbrainz_enabled(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_AUDIO_ROOT:
+            self._spawn(
+                self._admin_exec_audio_root(pending, transcript, sig_bytes))
         elif pending["op"] == OP_ROOT_ADD:
             self._spawn(
                 self._admin_exec_root_add(pending, transcript, sig_bytes))