Merge branch 'refactor/groups-phase1'

Groups refactor, phases 1-3.

The root model replaces the old `upload` flag and group-wide `member_upload`
with per-root `writable`/`removable`/`ejected`, carried by a `RootSet` that
both front doors — the loopback API and signed MNP — reach through the same
`ops` functions. MNP goes to 1.1, additively: the roots table now rides on
`index_delta`, so a root added, removed, ejected or plugged reaches every
connected client instead of only whoever reloaded.

The group UI becomes a plugin architecture: an application is a registry
entry in `apps.js` plus its own files, with directories stored generically
by `ops.set_app_directories` under whatever the app is called. A reference
application, hidden behind `?dev=1`, is what makes that claim testable —
adding it is what found the two places still naming apps by hand.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011pvMdvLBG92jyhvD5pD6us

33 files changed, 4167 insertions, 702 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index 0e5a7de..7673a51 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -89,25 +89,20 @@ transcode_incompatible_video = true
 # A root's name is the directory's basename, and it becomes the first segment of
 # every path members see: /home/user/Media appears to everyone as "Media/".
 # Two roots cannot share a name (compared without regard to case), and no root
-# may sit inside another. Exactly one root receives uploads.
+# may sit inside another. A writable root accepts uploads from group members.
 [[groups]]
 id          = ""    # set after joining
 name        = "My Media"
 quic_port   = 19010
 
   [[groups.roots]]
-  path   = "/home/user/Media"
-  upload = true
+  path     = "/home/user/Media"
+  writable = true
 
   [[groups.roots]]
-  path = "/run/media/user/USB/Musique"   # an external drive is fine: if it is
-  kind = "audio"                         # unplugged the root goes unavailable
-                                         # and its files stay in the index,
-                                         # rather than looking deleted
-
-# upload_dir: a separate directory for uploads. Files land directly in it,
-# not in an "uploads" subdirectory. It appears as its own root in the index.
-# upload_dir = "/home/user/Incoming"
+  path      = "/run/media/user/USB/Musique"
+  kind      = "audio"
+  removable = true   # eject before unplugging
 
 # The single-directory form still works and means the same thing — one root,
 # named after the directory, receiving uploads.
@@ -187,8 +182,8 @@ class RootSpec:
     path:   str = ""
     name:   str = ""          # empty → the directory's basename, derived at load
     kind:   str = "generic"   # generic|video|audio|photo — a view hint, unused for now
-    upload: bool = False      # exactly one root per group receives uploads
-    direct: bool = False      # uploads land at root path, not in a subdirectory
+    writable:  bool = False   # RW roots accept uploads from group members
+    removable: bool = False   # operator can eject this root before unplugging the device
 
 
 @dataclass
@@ -201,7 +196,7 @@ class GroupConfig:
     # unprefixed shape.
     roots:      list[RootSpec] = field(default_factory=list)
     shared_dir: str = ""      # legacy single-root form, migrated at load
-    upload_dir: str = ""      # separate filesystem path for uploads
+    upload_dir: str = ""      # legacy — migrated to a writable root
     visibility: str = "private"   # public|private — discoverability, not admission
     # Admission. "invite" (default) means a newcomer needs a one-time pairing code
     # before the node wraps the group key for them; "open" means the node pins
@@ -224,12 +219,12 @@ class GroupConfig:
         which reads like configuration rather than a bug.
         """
         if not self.roots and self.shared_dir.strip():
-            self.roots = [RootSpec(path=self.shared_dir.strip(), upload=True)]
+            self.roots = [RootSpec(path=self.shared_dir.strip(), writable=True)]
         if self.upload_dir.strip():
             for r in self.roots:
-                r.upload = False
+                r.writable = False
             self.roots.append(RootSpec(
-                path=self.upload_dir.strip(), upload=True, direct=True))
+                path=self.upload_dir.strip(), writable=True))
 
 
 @dataclass
@@ -285,15 +280,17 @@ def _read_roots(group: dict) -> list[RootSpec]:
     than merged: which one receives uploads would be a guess, and a wrong guess
     is discovered weeks later.
     """
-    specs = [
-        RootSpec(
+    specs = []
+    for r in group.get("roots", []) or []:
+        # Backward compat: old configs have `upload = true` instead of `writable`
+        writable = bool(r.get("writable", r.get("upload", False)))
+        specs.append(RootSpec(
             path=str(r.get("path", "")),
             name=str(r.get("name", "")),
             kind=str(r.get("kind", "generic")),
-            upload=bool(r.get("upload", False)),
-        )
-        for r in group.get("roots", []) or []
-    ]
+            writable=writable,
+            removable=bool(r.get("removable", False)),
+        ))
     legacy = str(group.get("shared_dir", "") or "").strip()
     if specs and legacy:
         log.warning(
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index ea13680..f9e992c 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -36,6 +36,7 @@ from pathlib import Path
 
 import uvicorn
 
+from meshbay_common.paths import fold
 from meshbay_common import MNP_VERSION
 from meshbay_common.protocol import MNP
 from meshbay_node.audit import AuditStore
@@ -69,26 +70,36 @@ if WEBRTC_AVAILABLE:
 log = logging.getLogger(__name__)
 
 
-def _under_video_root(path: str, video_root: str) -> bool:
-    """Mirrors video-app.js's underVideoRoot: same folder, or a descendant."""
-    path = path or ""
-    return path == video_root or path.startswith(video_root + "/")
-
+def _under_any_directory(path: str, directories: list[str]) -> bool:
+    """
+    Whether an entry's folder is one of an app's directories, or inside one.
 
-def _under_audio_root(path: str, audio_root: str) -> bool:
-    """Mirrors music-app.js's underAudioRoot — same shape as _under_video_root."""
+    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 path == audio_root or path.startswith(audio_root + "/")
+    return any(path == d or path.startswith(d + "/") for d in directories)
 
 
-def _under_any_photo_root(path: str, photo_roots: list[str]) -> bool:
+def _owning_directory(path: str, directories: list[str]) -> str | None:
     """
-    Mirrors photos-app.js's underAnyPhotoRoot. Unlike video/audio's single
-    root, photo_roots is a list (docs/photos.md §2.1) — a match against any
-    one of them is enough.
+    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 ""
-    return any(path == r or path.startswith(r + "/") for r in photo_roots)
+    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 ──────────────────────────────────────────────────────
@@ -128,6 +139,11 @@ class _WsSender:
 
 # ── 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
@@ -303,7 +319,7 @@ class NodeDaemon:
                     continue
 
                 try:
-                    roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                    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.
@@ -330,7 +346,7 @@ class NodeDaemon:
                     log.info("No GEK yet for group %s — will accept first setup",
                              group_cfg.name)
 
-                # Read once at load, like member_upload/enabled_apps below —
+                # 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.
@@ -347,6 +363,7 @@ class NodeDaemon:
                     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"],
@@ -374,37 +391,30 @@ class NodeDaemon:
                     "note_activity": indexer.note_activity,
                     # Shown to the operator in Settings, and kept current in
                     # place by set_scan_settings (ops.py) — same reasoning as
-                    # member_upload below.
+                    # 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,
-                    # Whether ordinary members may upload. Read once here, into
-                    # the context, because the upload handler is synchronous and
-                    # a database round trip per chunk would be absurd. The
-                    # signed operation that changes it updates this dict in
-                    # place, so the two never drift within a run.
-                    "member_upload": await self._roster.member_upload_allowed(
-                        group_cfg.id) if self._roster else True,
-                    # Same reasoning: read once at load, kept current in place
-                    # by the signed operation that changes it.
+                    # 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),
-                    # Which folder is the Videos app's entry point for this
-                    # group — "" means the whole group index.
-                    "video_root": await self._roster.video_root(
-                        group_cfg.id) if self._roster else "",
-                    # Same shape, Music app's own entry point.
-                    "audio_root": await self._roster.audio_root(
-                        group_cfg.id) if self._roster else "",
-                    # Photos app's entry points — a *list*, unlike video_root/
-                    # audio_root above (docs/photos.md §2.1: a photo library
-                    # is routinely scattered across several folders). Empty
-                    # list means nothing configured yet.
-                    "photo_roots": await self._roster.photo_roots(
-                        group_cfg.id) if self._roster else [],
+                    # 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,
                     # 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"
@@ -627,9 +637,14 @@ class NodeDaemon:
             self._state["quic_server"] = self._quic_server
             self._state["hub"] = hub
             self._state["reload_fn"] = self._reload_config
-            self._state["enrich_video_root_fn"] = self._enrich_video_root_now
-            self._state["enrich_audio_root_fn"] = self._enrich_audio_root_now
-            self._state["enrich_photo_roots_fn"] = self._enrich_photo_roots_now
+            # 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
@@ -745,14 +760,16 @@ class NodeDaemon:
             if not ctx:
                 continue
             try:
-                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                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
-            before = {(r.name, str(r.path)) for r in ctx["roots"]}
-            after = {(r.name, str(r.path)) for r in roots}
-            if before == after:
+            # `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
@@ -786,7 +803,7 @@ class NodeDaemon:
                 continue
 
             try:
-                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                roots = await self._build_roots(group_cfg)
             except RootError as e:
                 log.error("New group %r: %s — skipping", group_cfg.name, e)
                 continue
@@ -812,6 +829,7 @@ class NodeDaemon:
                 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"],
@@ -844,21 +862,13 @@ class NodeDaemon:
                 "debounce_secs": scan_settings["debounce_secs"],
                 "visibility": group_cfg.visibility,
                 "join_policy": group_cfg.join_policy,
-                "member_upload": (
-                    await self._roster.member_upload_allowed(group_cfg.id)
-                    if self._roster else True),
                 "enabled_apps": (
                     await self._roster.enabled_apps(group_cfg.id)
                     if self._roster else list(Roster.DEFAULT_APPS)),
-                "video_root": (
-                    await self._roster.video_root(group_cfg.id)
-                    if self._roster else ""),
-                "audio_root": (
-                    await self._roster.audio_root(group_cfg.id)
-                    if self._roster else ""),
-                "photo_roots": (
-                    await self._roster.photo_roots(group_cfg.id)
-                    if self._roster else []),
+                **(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),
                 "tmdb_enabled": (
                     await self._roster.tmdb_enabled(group_cfg.id)
                     if self._roster else True),
@@ -1052,6 +1062,65 @@ class NodeDaemon:
             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 app that keeps directories. Not derived from `enabled_apps`: the
+    # context is read once at load and an app enabled later must not find its
+    # own setting missing. Adding an app adds a name here and nowhere else on
+    # this side.
+    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 legacy scalar names the
+        rest of the tree still reads.
+
+        The scalars are derived here rather than stored, so the two can never
+        disagree: `video_root` is the first of `video_directories` and exists
+        for MNP 1.0 clients and for the handful of call sites that predate the
+        list. A group with several video directories reports the first as its
+        `video_root` — which is what an old client can represent, and all it
+        could ever have shown.
+        """
+        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
@@ -1103,6 +1172,24 @@ class NodeDaemon:
         # 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]
         asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries))
         # Music app (docs/musicbay.md §6): same shape, gated on audio_root
         # exactly like video_root above (added later — musicbay.md's
@@ -1171,7 +1258,8 @@ class NodeDaemon:
             # 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) if delta is not None
+                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:
@@ -1216,13 +1304,13 @@ class NodeDaemon:
         """
         if not self._enricher or not self._roster:
             return
-        video_root = await self._roster.video_root(indexer.group_id)
-        if not video_root:
+        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_video_root(entry.path, video_root):
+            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():
@@ -1297,14 +1385,18 @@ class NodeDaemon:
         """
         if not self._audio_enricher or not self._roster:
             return
-        audio_root = await self._roster.audio_root(indexer.group_id)
-        if not audio_root:
+        audio_dirs = await self._roster.app_directories(indexer.group_id, "music")
+        if not audio_dirs:
             return
-        root_boundary = indexer.roots.resolve(audio_root, require_available=False)
+        # 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
-            if not _under_audio_root(entry.path, audio_root):
+            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():
@@ -1314,13 +1406,16 @@ class NodeDaemon:
             async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None:
                 await self._on_enriched(_indexer, file_id, fields)
 
-            # `root_boundary` — audio_root itself, not the shared root it
-            # lives under — so the ancestor walk
+            # 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 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)
+            # 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:
         """
@@ -1370,13 +1465,13 @@ class NodeDaemon:
         """
         if not self._photo_enricher or not self._roster:
             return
-        photo_roots = await self._roster.photo_roots(indexer.group_id)
-        if not photo_roots:
+        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_photo_root(entry.path, photo_roots):
+            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():
@@ -1663,15 +1758,17 @@ def main() -> None:
     parser = argparse.ArgumentParser(description="MeshBay Node daemon")
     parser.add_argument("command", nargs="?",
                         choices=["init", "reset", "status", "gek-init",
-                                 "gek", "operator", "member", "group", "file",
-                                 "video", "denylist", "stun", "reload",
+                                 "gek", "operator", "member", "group", "root",
+                                 "file", "video", "denylist", "stun", "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 | gek init|rotate | file list|rm "
+                             "| 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 | denylist show|clear "
                              "| stun list|add|remove|reset "
@@ -1687,7 +1784,9 @@ def main() -> None:
                              "| calibrate-argon2: benchmark")
     parser.add_argument("subcommand", nargs="?",
                         help="'pair' for operator; list|invite|revoke|unpin for "
-                             "member; list|add|remove for group; init|rotate for gek; "
+                             "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; "
                              "install|remove|start|stop|status for autostart and "
@@ -1702,22 +1801,35 @@ def main() -> None:
                         help="hub username, for init")
     parser.add_argument("--dir", default=None,
                         help="shared directory, for group add")
-    parser.add_argument("--upload-dir", default=None,
-                        help="separate upload directory, for group add")
     parser.add_argument("--yes", action="store_true",
                         help="skip the confirmation for destructive commands")
     parser.add_argument("--config", type=Path, default=None,
                         help="Config file path")
     parser.add_argument("--group", default=None,
                         help="group id (optional if only one is configured)")
+    parser.add_argument("--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", "file", "video", "denylist",
-                             "stun", "reload", "restart-daemon", "reset")
+                             "member", "group", "root", "file", "video",
+                             "denylist", "stun", "reload", "restart-daemon",
+                             "reset")
     logging.basicConfig(
         level=logging.ERROR if quiet else getattr(logging, args.log_level),
         format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
@@ -1961,9 +2073,16 @@ def main() -> None:
                     print("             <no directory configured>")
                 for r in g.roots:
                     label = r.name or Path(r.path).name
-                    flag = "  (uploads)" if r.upload else ""
+                    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}{live}")
+                    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.
@@ -2030,6 +2149,26 @@ def main() -> None:
                           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)
@@ -2408,11 +2547,18 @@ def main() -> None:
                       f"{g.get('peers', 0)} peer(s)")
                 print(f"      {g['id']}")
                 for r in g.get("roots", []):
-                    flags = ""
-                    if r.get("upload"):
-                        flags = " (uploads, direct)" if r.get("direct") else " (uploads)"
+                    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']}{flags}{live}")
+                    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']}")
@@ -2440,22 +2586,27 @@ def main() -> None:
             print("usage: meshbay-node group list|add|remove <name>")
             sys.exit(1)
         if not args.target or not args.dir:
-            print("usage: meshbay-node group add <name> --dir <path> [--upload-dir <path>]")
+            print("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 the directory.")
-            print("--upload-dir sets a separate directory for uploaded files.")
+            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)
-        body = {"name": args.target, "shared_dir": args.dir}
-        if args.upload_dir:
-            body["upload_dir"] = args.upload_dir
+        # 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']}")
-        if out.get("upload_dir"):
-            print(f"  upload_dir   {out['upload_dir']}")
+        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")
@@ -2465,6 +2616,124 @@ def main() -> None:
         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")
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index b0a8e50..33e7210 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -263,12 +263,17 @@ class DirectoryIndexer:
         cache: IndexCache | None = None,
         reconcile_secs: float = DEFAULT_RECONCILE_SECS,
         debounce_secs: float = DEFAULT_DEBOUNCE_SECS,
+        on_root_ejected: Callable[[str, bool], Awaitable[None]] | None = None,
     ):
         self.roots    = roots
         self.group_id = group_id
         self.sk_node  = sk_node
         self.gek      = gek
         self.on_change = on_change
+        # Called with (root_name, ejected) whenever this indexer changes a
+        # root's ejected state by itself — the surprise-unplug safety net.
+        # The daemon writes it to the roster, so a restart does not undo it.
+        self.on_root_ejected = on_root_ejected
         self.reconcile_secs = reconcile_secs
         self.debounce_secs = debounce_secs
         # Current backoff delay — starts at reconcile_secs, doubles on every
@@ -309,6 +314,14 @@ class DirectoryIndexer:
         # stay up for exactly as long as the slow part (hashing) is running.
         self._burst_inflight = 0
         self._burst_sizes: dict[str, int] = {}
+        # Ids whose entry this indexer threw away and rebuilt from disk, since
+        # the last time a consumer drained this. A rebuilt entry carries only
+        # what `_hash_or_cached` fills in — every enrichment field the Videos,
+        # Music and Photos apps put there is gone — but its id is the file's
+        # content hash, so a diff against the last broadcast sees no addition
+        # and no deletion and nothing downstream can tell the fields were
+        # wiped. See `_drop_root_entries`.
+        self.rescanned_ids: set[str] = set()
 
     @property
     def index(self) -> GroupIndex:
@@ -582,11 +595,21 @@ class DirectoryIndexer:
         changed = self.roots.refresh_availability()
         touched = False
 
+        # Drained before the loop below, because persisting the flag is what
+        # makes the safety net survive a restart — and a restart is exactly
+        # what an operator does after noticing a drive fell off.
+        while self.roots.auto_ejected:
+            name = self.roots.auto_ejected.pop(0)
+            if self.on_root_ejected:
+                try:
+                    await self.on_root_ejected(name, True)
+                except Exception:
+                    log.exception("Could not persist the auto-eject of root %r", name)
+
         for root, available in changed:
             if available:
                 log.info("Root %r is back — rescanning", root.name)
-                self._drop_root_entries(root)
-                await self._scan_root(root)
+                await self._rescan_root(root)
                 touched = True
             else:
                 # Frozen: entries stay, marked unavailable to members through
@@ -687,10 +710,71 @@ class DirectoryIndexer:
         return [e for e in self._index.entries
                 if fold(e.path).split("/", 1)[0] == prefix]
 
+    # Everything on an IndexEntry that a scan does not produce. `_scan_root`
+    # fills id/name/path/size/type/added_at/hash_version from the file itself;
+    # every field below was derived by one of the enrichment passes and is
+    # nowhere on disk to be read back.
+    _ENRICHED_FIELDS = (
+        "duration", "thumb_hash", "width", "height",
+        "display_title", "season", "episode",
+        "artist", "album", "track_no", "taken_at", "camera",
+        "uploader_id", "uploader_pk",
+    )
+
+    async def _rescan_root(self, root: Root) -> int:
+        """
+        Rebuild one root's entries from disk, keeping what the files still say.
+
+        The two callers — `reconcile` when a root reappears, `plug_root` when
+        the operator plugs one back in — have to re-walk: the drive may have
+        changed while it was away. What they must not do is throw away the
+        enrichment. An entry's id is its content hash, so an entry that comes
+        back under the same id, name and path is the same bytes in the same
+        place, and every field the Videos, Music and Photos passes derived from
+        it still holds. Re-deriving them means minutes of tag reads, ffprobe
+        runs and rate-limited metadata lookups during which the operator's
+        library sits empty — which is exactly what a replug looked like.
+
+        Anything that does *not* match is left bare on purpose: a different id
+        is different content, and a different name or path can change the
+        filename and folder fallbacks that `display_title`, `track_no`,
+        `artist` and `album` fall back to. Those are the entries
+        `daemon._broadcast_index_change` re-enriches, off `rescanned_ids`.
+        """
+        carried = {(e.id, e.name, e.path): e for e in self._entries_under(root)}
+        self._drop_root_entries(root)
+        count = await self._scan_root(root)
+        for entry in self._entries_under(root):
+            old = carried.get((entry.id, entry.name, entry.path))
+            if old is None:
+                continue
+            for field in self._ENRICHED_FIELDS:
+                setattr(entry, field, getattr(old, field))
+            # It came back intact, so it is not one of the entries the daemon
+            # needs to enrich again.
+            self.rescanned_ids.discard(entry.id)
+        return count
+
     def _drop_root_entries(self, root: Root) -> None:
+        """
+        Throw away a root's entries. Only ever called to rebuild them.
+
+        The ids are recorded because nothing outside can otherwise tell they
+        were rebuilt: no deletion is broadcast (the rescan is immediate) and
+        the entries come back under the same content-hash ids, so a diff
+        against the last broadcast reports neither an addition nor a deletion.
+        `_rescan_root` clears the ones it managed to carry over intact; what is
+        left is genuinely new to the apps and is re-enriched by the daemon.
+        """
         for entry in self._entries_under(root):
+            self.rescanned_ids.add(entry.id)
             self._index.remove_entry(entry.id)
 
+    def drain_rescanned_ids(self) -> set[str]:
+        """Take the ids rebuilt since the last call; leave the set empty."""
+        drained, self.rescanned_ids = self.rescanned_ids, set()
+        return drained
+
     @staticmethod
     def _entry_path(root: Root, entry: IndexEntry) -> Path | None:
         _, _, tail = entry.path.partition("/")
@@ -723,6 +807,43 @@ class DirectoryIndexer:
             self._observer = None
         self._start_observer()
 
+    def eject_root(self, root_name: str) -> None:
+        """Stop watching a root without touching its entries."""
+        from meshbay_common.paths import fold
+        target = fold(root_name)
+        for root in self.roots:
+            if fold(root.name) == target:
+                root.ejected = True
+                root.available = False
+                frozen = len(self._entries_under(root))
+                log.info("Root %r ejected — %d entries frozen", root.name, frozen)
+                break
+        self._restart_observer()
+        self._index.roots = self.roots.describe()
+        self._index.version = int(time.time())
+
+    async def plug_root(self, root_name: str) -> None:
+        """Restart watching a previously ejected root and reconcile."""
+        from meshbay_common.paths import fold
+        target = fold(root_name)
+        root = None
+        for r in self.roots:
+            if fold(r.name) == target:
+                root = r
+                break
+        if root is None:
+            return
+        root.ejected = False
+        root.available = root.is_live()
+        if root.available:
+            log.info("Root %r plugged — rescanning", root.name)
+            await self._rescan_root(root)
+        self._restart_observer()
+        self._index.roots = self.roots.describe()
+        self._index.version = int(time.time())
+        if self.on_change:
+            await self.on_change(self)
+
     # ── Internal update ───────────────────────────────────────────────────────
 
     def _schedule_update(self, file_path: Path, deleted: bool = False) -> None:
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
index 4c20c2a..a10504e 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -362,7 +362,11 @@ async def list_groups(state: dict) -> dict:
             "has_gek": bool(ctx.get("gek")),
             "file_count": idx.count if idx else 0,
             "index_version": idx.version if idx else 0,
-            "roots": roots.describe() if roots else [],
+            # With paths: this answers the loopback API, which is the
+            # operator's own channel. `meshbay-node root list` printed "?" for
+            # every directory without it — it was reading a field the member
+            # form of this deliberately omits.
+            "roots": roots.describe(with_paths=True) if roots else [],
             "peers": sum(1 for p in peers.values() if p.get("group_id") == gid),
         })
     roster = state.get("roster")
@@ -390,7 +394,7 @@ async def list_groups(state: dict) -> dict:
 
 
 async def attach_group(state: dict, name: str, shared_dir: str,
-                       upload_dir: str = "") -> dict:
+                       writable: bool = True) -> dict:
     """
     Write a new [[groups]] block into node.toml.
 
@@ -429,31 +433,24 @@ async def attach_group(state: dict, name: str, shared_dir: str,
         raise OpError(f"Cannot create {path}: {e}") from e
 
     conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
-    # Appended as text rather than re-serialised: node.toml is hand-written and
-    # full of comments explaining decisions, and a round trip through a TOML
-    # writer would throw all of that away.
     join_policy = group.get("join_policy", "invite")
     block = (f'\n[[groups]]\n'
              f'id          = "{group["id"]}"\n'
              f'name        = "{group["name"]}"\n'
              f'visibility  = "{group.get("visibility", "private")}"\n'
              f'join_policy = "{join_policy}"\n')
-    separate_upload = False
-    if upload_dir:
-        upload_path = Path(upload_dir).expanduser().resolve()
-        if upload_path != path.resolve():
-            separate_upload = True
-            try:
-                upload_path.mkdir(parents=True, exist_ok=True)
-            except OSError as e:
-                raise OpError(f"Cannot create {upload_path}: {e}") from e
-            block += f'upload_dir  = "{upload_path.as_posix()}"\n'
+    # No `upload_dir` here. `GroupConfig.__post_init__` still *reads* it, so an
+    # existing node.toml keeps working — but what it does on read is force every
+    # other root read-only and append that path as the one writable one, which
+    # is the model this refactor replaced. Writing it into a group created
+    # today would mean two mechanisms deciding the same thing, one of them
+    # invisible: `group add --dir X --writable --upload-dir Y` silently made X
+    # read-only. A second writable directory is `root add <path> --writable`.
     block += (f'\n  [[groups.roots]]\n'
               # Forward slashes: a Windows path in a TOML basic string is a
               # parse error (`\U`, `\a`, ... are escapes). pathlib reads `/`.
-              f'  path   = "{path.as_posix()}"\n')
-    if not separate_upload:
-        block += f'  upload = true\n'
+              f'  path     = "{path.as_posix()}"\n'
+              f'  writable = {"true" if writable else "false"}\n')
     try:
         with conf_path.open("a", encoding="utf-8", newline="\n") as f:
             f.write(block)
@@ -462,9 +459,8 @@ async def attach_group(state: dict, name: str, shared_dir: str,
 
     result = {"group_id": group["id"], "name": group["name"],
               "shared_dir": str(path), "config": str(conf_path),
+              "writable": writable,
               "note": "restart the node to pick it up"}
-    if separate_upload:
-        result["upload_dir"] = str(upload_path)
     return result
 
 
@@ -636,12 +632,13 @@ def _remove_roots_block(conf_path: Path, group_id: str,
                     conf_path.write_text("\n".join(new_lines), encoding="utf-8", newline="\n")
                     return
 
-    raise OpError(f"Root path not found in config", status=404)
+    raise OpError("Root path not found in config", status=404)
 
 
 async def add_root(state: dict, group_id: str, path: str, *,
                    name: str = "", kind: str = "generic",
-                   upload: bool = False) -> dict:
+                   writable: bool = False,
+                   removable: bool = False) -> dict:
     """
     Add a directory to a group, refusing anything ambiguous.
 
@@ -655,7 +652,8 @@ async def add_root(state: dict, group_id: str, path: str, *,
         raise OpError("Group not configured on this node", status=404)
 
     specs = [asdict(r) for r in cfg.roots]
-    specs.append({"path": path, "name": name, "kind": kind, "upload": upload})
+    specs.append({"path": path, "name": name, "kind": kind,
+                  "writable": writable, "removable": removable})
     try:
         built = RootSet.build(specs)
     except RootError as e:
@@ -674,15 +672,29 @@ async def add_root(state: dict, group_id: str, path: str, *,
         root_block += f'\n  name   = "{added.name}"'
     if kind != "generic":
         root_block += f'\n  kind   = "{added.kind}"'
-    if upload:
-        root_block += f'\n  upload = true'
+    if writable:
+        root_block += '\n  writable = true'
+    if removable:
+        root_block += '\n  removable = true'
     _insert_roots_block(conf_path, group_id, root_block)
 
     from meshbay_node.config import RootSpec
     cfg.roots.append(RootSpec(
         path=str(added.path), name=added.name, kind=added.kind,
-        upload=added.upload, direct=added.direct))
+        writable=added.writable, removable=added.removable))
 
+    # Deliberately *not* mutating the live RootSet in place.
+    #
+    # `DirectoryIndexer.retarget` decides what to scan by diffing the names it
+    # already has against the ones it is given — so handing it the same object,
+    # edited, means the new root is in both sides of the comparison and is
+    # never scanned. It would appear in the table and stay permanently empty.
+    # `_reload_config_inner` diffs the same way and would likewise conclude
+    # nothing changed. The caller reloads instead, which builds a fresh set
+    # from the file this just wrote.
+    #
+    # `built` is that set, computed here only to validate and to answer with;
+    # what the node serves comes from the reload.
     log.info("Root added: %s → group %s", added.name, group_id[:8])
     return {"status": "added", "name": added.name, "path": str(added.path),
             "group_id": group_id, "roots": built.describe()}
@@ -714,25 +726,245 @@ async def remove_root(state: dict, group_id: str, root_name: str) -> dict:
         raise OpError("Cannot remove the only root", status=400)
 
     removed = cfg.roots[match_idx]
-    if removed.upload:
-        raise OpError(
-            "Cannot remove the upload root — file uploads and chat "
-            "attachments are stored there", status=400)
     resolved = str(Path(removed.path).expanduser().resolve())
 
     conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
     _remove_roots_block(conf_path, group_id, resolved)
 
     cfg.roots.pop(match_idx)
-    remaining = [asdict(r) for r in cfg.roots]
-    try:
-        built = RootSet.build(remaining)
-    except RootError:
-        built = None
+
+    # Not mutating the live set here either — see `add_root`. Dropping the
+    # root from it would leave `retarget` unable to tell that its entries
+    # should go, so the removed directory's files would stay in the index.
+    #
+    # Built from the config this just edited, and never returned empty: an
+    # empty list is a *valid answer* meaning "this group has no directories",
+    # which the client cannot tell from "the node could not say" — it would
+    # blank the operator's table on an op that succeeded.
+    result_roots = RootSet.build([asdict(r) for r in cfg.roots]).describe()
 
     log.info("Root removed: %s from group %s", root_name, group_id[:8])
     return {"status": "removed", "name": root_name, "group_id": group_id,
-            "roots": built.describe() if built else []}
+            "roots": result_roots}
+
+
+async def update_root(state: dict, group_id: str, root_name: str, *,
+                      writable: bool | None = None,
+                      removable: bool | None = None) -> dict:
+    """Toggle writable/removable on an existing root without removing it."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+    from meshbay_node.roots import RootSet
+    target = fold(root_name)
+    match = None
+    for r in cfg.roots:
+        rname = r.name or str(Path(r.path).name)
+        if fold(rname) == target:
+            match = r
+            break
+    if match is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+
+    changed = False
+    if writable is not None and match.writable != writable:
+        match.writable = writable
+        changed = True
+    if removable is not None and match.removable != removable:
+        match.removable = removable
+        changed = True
+
+    if not changed:
+        specs = [asdict(r) for r in cfg.roots]
+        built = RootSet.build(specs)
+        return {"status": "unchanged", "name": root_name, "group_id": group_id,
+                "roots": built.describe()}
+
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    _update_root_field(conf_path, group_id, str(Path(match.path).expanduser().resolve()),
+                       writable=match.writable, removable=match.removable)
+
+    # Update the live RootSet so GET /api/groups returns correct data
+    # immediately, without waiting for the async reload to finish.
+    live_roots: RootSet | None = state.get("groups_ctx", {}).get(
+        group_id, {}).get("roots")
+    if live_roots:
+        for lr in live_roots.roots:
+            lr_name = lr.name or str(Path(lr.path).name)
+            if fold(lr_name) == target:
+                if writable is not None:
+                    lr.writable = writable
+                if removable is not None:
+                    lr.removable = removable
+                break
+
+    # Built from config when there is no live set, never returned empty: an
+    # empty list is a *valid answer* meaning "this group has no directories",
+    # and the client cannot tell it from "the node could not say". It would
+    # blank the operator's table on an op that succeeded.
+    result_roots = (live_roots.describe() if live_roots
+                    else RootSet.build([asdict(r) for r in cfg.roots]).describe())
+
+    log.info("Root updated: %s (writable=%s, removable=%s) in group %s",
+             root_name, match.writable, match.removable, group_id[:8])
+    return {"status": "updated", "name": root_name, "group_id": group_id,
+            "roots": result_roots}
+
+
+async def eject_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Mark a removable root as ejected so the operator can safely unplug."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+    target = fold(root_name)
+    ctx = _group_ctx(state, group_id)
+    roots: RootSet | None = ctx.get("roots")
+    if not roots:
+        raise OpError("Group has no roots", status=503)
+
+    root = None
+    for r in roots:
+        if fold(r.name) == target:
+            root = r
+            break
+    if root is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+    if not root.removable:
+        raise OpError(f"Root {root_name!r} is not marked as removable", status=400)
+    if root.ejected:
+        return {"status": "already_ejected", "name": root_name,
+                "group_id": group_id, "roots": roots.describe()}
+
+    # The indexer stops its watchdog and freezes the entries; it holds the same
+    # RootSet object, but the flags are set here too so a context whose indexer
+    # was replaced by a retarget cannot be left disagreeing with the roster.
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        indexer.eject_root(root_name)
+    root.ejected = True
+    root.available = False
+
+    await _roster(state).set_root_ejected(
+        group_id, root_name, True, set_by=state.get("node_user_id", ""))
+
+    log.info("Root ejected: %s from group %s", root_name, group_id[:8])
+    return {"status": "ejected", "name": root_name, "group_id": group_id,
+            "roots": roots.describe()}
+
+
+async def plug_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Re-enable an ejected root after the device is plugged back in."""
+    config = _config(state)
+    cfg = next((g for g in config.groups if g.id == group_id), None)
+    if cfg is None:
+        raise OpError("Group not configured on this node", status=404)
+
+    from meshbay_common.paths import fold
+    target = fold(root_name)
+    ctx = _group_ctx(state, group_id)
+    roots: RootSet | None = ctx.get("roots")
+    if not roots:
+        raise OpError("Group has no roots", status=503)
+
+    root = None
+    for r in roots:
+        if fold(r.name) == target:
+            root = r
+            break
+    if root is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+    if not root.ejected:
+        return {"status": "already_plugged", "name": root_name,
+                "group_id": group_id, "roots": roots.describe()}
+    if not root.is_live():
+        raise OpError(
+            f"Directory not found: {root.path}. Is the device connected?",
+            status=409)
+
+    # Persisted before the rescan, which can take minutes on a large library:
+    # a crash halfway through must leave the root plugged, not ejected with
+    # entries half rebuilt.
+    await _roster(state).set_root_ejected(
+        group_id, root_name, False, set_by=state.get("node_user_id", ""))
+
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        await indexer.plug_root(root_name)
+    root.ejected = False
+    root.available = root.is_live()
+
+    log.info("Root plugged: %s in group %s", root_name, group_id[:8])
+    return {"status": "plugged", "name": root_name, "group_id": group_id,
+            "roots": roots.describe()}
+
+
+def _update_root_field(conf_path: Path, group_id: str,
+                       resolved_path: str, *,
+                       writable: bool, removable: bool) -> None:
+    """Update writable/removable fields on a root in node.toml."""
+    text = conf_path.read_text(encoding="utf-8")
+    lines = text.split("\n")
+
+    rng = _find_group_range(lines, group_id)
+    if rng is None:
+        raise OpError(f"Group {group_id[:8]} not found in {conf_path}")
+
+    start, end = rng
+    path_re = re.compile(r'^\s*path\s*=\s*"([^"]*)"')
+    writable_re = re.compile(r'^\s*(writable|upload)\s*=')
+    removable_re = re.compile(r'^\s*removable\s*=')
+    roots_starts: list[int] = []
+    for i in range(start + 1, end):
+        if lines[i].strip() == "[[groups.roots]]":
+            roots_starts.append(i)
+
+    for j, rs in enumerate(roots_starts):
+        rs_end = roots_starts[j + 1] if j + 1 < len(roots_starts) else end
+        found_path = False
+        for k in range(rs, rs_end):
+            m = path_re.match(lines[k])
+            if m:
+                try:
+                    p = str(Path(m.group(1)).expanduser().resolve())
+                except OSError:
+                    continue
+                if p == resolved_path:
+                    found_path = True
+                    break
+        if not found_path:
+            continue
+
+        writable_idx = None
+        removable_idx = None
+        for k in range(rs, rs_end):
+            if writable_re.match(lines[k]):
+                writable_idx = k
+            if removable_re.match(lines[k]):
+                removable_idx = k
+
+        if writable_idx is not None:
+            lines[writable_idx] = f"  writable = {'true' if writable else 'false'}"
+        else:
+            lines.insert(rs_end, f"  writable = {'true' if writable else 'false'}")
+            if removable_idx is not None and removable_idx >= rs_end:
+                removable_idx += 1
+            rs_end += 1
+
+        if removable_idx is not None:
+            lines[removable_idx] = f"  removable = {'true' if removable else 'false'}"
+        else:
+            lines.insert(rs_end, f"  removable = {'true' if removable else 'false'}")
+
+        conf_path.write_text("\n".join(lines), encoding="utf-8", newline="\n")
+        return
+
+    raise OpError("Root path not found in config", status=404)
 
 
 # ── Files ────────────────────────────────────────────────────────────────────
@@ -806,26 +1038,6 @@ async def clear_denylist(state: dict, *, subject: str = "") -> dict:
     return {"status": "cleared", "removed": removed, "subject": subject or "all"}
 
 
-# ── Upload policy ───────────────────────────────────────────────────────────
-
-async def set_member_upload(state: dict, group_id: str, allowed: bool) -> dict:
-    """
-    Turn uploading by ordinary members on or off.
-
-    The setting lives on the node (roster.db), not on the hub and not in
-    node.toml — changing it must not rewrite the operator's config file,
-    and must not need a restart.
-    """
-    roster = _roster(state)
-    ctx = _group_ctx(state, group_id)
-    await roster.set_member_upload(group_id, allowed,
-                                   set_by=state.get("node_user_id", ""))
-    ctx["member_upload"] = allowed
-    log.info("Upload policy: %s for group %s", "on" if allowed else "off",
-             group_id[:8])
-    return {"allowed": allowed, "group_id": group_id}
-
-
 # ── Node settings ────────────────────────────────────────────────────────────
 
 async def get_node_settings(state: dict) -> dict:
@@ -924,12 +1136,16 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict:
     """
     Which group "applications" (Chat, Files, ...) are shown to members.
 
-    Same shape as `set_member_upload`: lives on the node (roster.db), takes
+    Same shape as other signed ops: lives on the node (roster.db), takes
     effect without a restart, and is signed by the operator (webrtc_server.py
     checks the caller's own admin-authority allow-list before this runs).
     """
     roster = _roster(state)
     ctx = _group_ctx(state, group_id)
+    # See the same guard in webrtc_server._do_apps_enabled: Files cannot be
+    # turned off, and both writers put it at the front so the two agree.
+    if "files" not in apps:
+        apps = ["files"] + list(apps)
     await roster.set_enabled_apps(group_id, apps,
                                   set_by=state.get("node_user_id", ""))
     ctx["enabled_apps"] = apps
@@ -946,7 +1162,7 @@ async def set_tmdb_config(state: dict, token: str | None = None,
     and in what language it queries TMDB (docs/mediacenter.md §5.5).
 
     Node-wide (roster.py group_settings, group_id="") rather than per-group
-    like set_member_upload/set_enabled_apps: the token and the shared-cache
+    like set_enabled_apps: the token and the shared-cache
     language are one operator's budget and one credential, not a per-group
     or per-viewer concern. Whether TMDB is used *at all* is the per-group
     decision set_tmdb_enabled below makes instead. `token=""` explicitly
@@ -1007,76 +1223,171 @@ async def set_musicbrainz_enabled(state: dict, group_id: str, enabled: bool) ->
     return {"enabled": enabled, "group_id": group_id}
 
 
-async def set_video_root(state: dict, group_id: str, path: str) -> dict:
+# ── App directories ──────────────────────────────────────────────────────────
+
+def _validate_app_dirs(state: dict, group_id: str, paths: list[str], *,
+                       require_writable: bool) -> list[str]:
     """
-    Which folder (possibly a subfolder of a shared root) is the Videos app's
-    entry point for this group. Same shape as set_enabled_apps: lives on the
-    node (roster.db), takes effect without a restart, signed by the operator.
-    `path=""` clears it — the Videos tab then asks for one to be chosen
-    before anything (including TMDB enrichment, docs/mediacenter.md §5.2)
-    runs, rather than defaulting to the whole shared index.
+    Every path an app is pointed at must live inside one of the group's roots.
+
+    The per-app setters this replaces validated nothing: a typo, or a path left
+    behind by a root that was removed, was stored and then quietly matched no
+    entry — an app showing an empty tab with no way to tell "misconfigured"
+    from "no files yet". Refusing at the point of setting is the only moment
+    the operator is present to be told.
 
-    A non-empty path fires (never awaits) a sweep of whatever that folder
-    already contains: the ordinary per-change enrichment path only ever
-    looks at files new since the last broadcast, so anything already sitting
-    in a folder before it became the video_root would otherwise never be
-    picked up.
+    Not `RootSet.resolve()`, deliberately: that also refuses a directory whose
+    root is currently *unavailable*, and an operator must be able to configure
+    a library on a drive they have unplugged. What is checked here is the
+    shape — inside a named root, no traversal — which does not change with
+    what happens to be mounted.
     """
-    roster = _roster(state)
-    ctx = _group_ctx(state, group_id)
-    await roster.set_video_root(group_id, path, set_by=state.get("node_user_id", ""))
-    ctx["video_root"] = path
-    log.info("Videos root for group %s: %r", group_id[:8], path)
-    if path:
-        enrich_fn = state.get("enrich_video_root_fn")
-        if enrich_fn:
-            asyncio.ensure_future(enrich_fn(group_id))
-    return {"path": path, "group_id": group_id}
+    roots: RootSet | None = _group_ctx(state, group_id).get("roots")
+    if roots is None:
+        raise OpError("Group has no roots", status=503)
 
+    clean: list[str] = []
+    for raw in paths:
+        path = str(raw or "").strip().strip("/")
+        if not path:
+            continue
+        if ".." in path.split("/"):
+            raise OpError(f"{path!r} is not a directory inside this group",
+                          status=400)
+        found = roots.split(path)
+        if found is None:
+            raise OpError(
+                f"{path!r} is not inside any of this group's shared "
+                f"directories", status=400,
+                extra={"available": roots.names})
+        root, _tail = found
+        if require_writable and not root.writable:
+            raise OpError(
+                f"{root.name!r} is read-only, and this setting needs a "
+                f"directory that accepts uploads", status=400)
+        clean.append(path)
+    return sorted(set(clean))
 
-async def set_audio_root(state: dict, group_id: str, path: str) -> dict:
+
+async def set_app_directories(state: dict, group_id: str, app_key: str,
+                              paths: list[str], *,
+                              require_writable: bool = False) -> 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.
+    Which folder(s) inside the group's shared roots an application works over.
+
+    One function for every app, keyed by the app's own name: adding an
+    application is a registry entry and a settings component, not another
+    near-identical op here. It replaces `set_video_root`, `set_audio_root` and
+    `set_photo_roots`, which differed only in the key they wrote and whether
+    they took a string or a list.
+
+    Empty means nothing configured, which every app reads as "show nothing
+    until an operator has chosen" — never "the whole group index". Pointing an
+    app at the whole library is a decision, not a default nobody made.
+
+    A change always fires (never awaits) a sweep of what the new directories
+    already contain: the ordinary per-change enrichment path only looks at
+    entries new since the last broadcast, so files already sitting in a folder
+    when it was chosen would otherwise never be picked up.
     """
     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}
+    clean = _validate_app_dirs(state, group_id, paths,
+                               require_writable=require_writable)
+    await roster.set_app_directories(group_id, app_key, clean,
+                                     set_by=state.get("node_user_id", ""))
+    ctx[f"{app_key}_directories"] = clean
+    # The scalar the handshake ack still publishes for MNP 1.0 clients is
+    # derived, and has to be re-derived here: leaving it behind would make the
+    # ack disagree with the list within a single run, and only until a restart
+    # — the shape of bug that reads as "it works after a restart".
+    from meshbay_node.roster import Roster
+    alias = Roster.ctx_alias(app_key, clean)
+    if alias:
+        ctx[alias[0]] = alias[1]
+    log.info("%s directories for group %s: %s", app_key, group_id[:8],
+             ", ".join(clean) or "(none)")
+
+    enrich = (state.get("enrich_app_dirs_fns") or {}).get(app_key)
+    if enrich:
+        asyncio.ensure_future(enrich(group_id))
+    return {"app": app_key, "directories": clean, "group_id": group_id}
+
+
+async def set_app_directory(state: dict, group_id: str, app_key: str,
+                            path: str, *,
+                            require_writable: bool = False) -> dict:
+    """
+    The single-directory form, for an app that only ever wants one.
+
+    Stored as a one-element list like every other app, because two storage
+    shapes for one idea is what made `video_root` (scalar) and `photo_roots`
+    (list) need separate ops, separate MNP messages and separate widgets to
+    say the same thing. `path=""` clears it.
+    """
+    result = await set_app_directories(
+        state, group_id, app_key, [path] if path else [],
+        require_writable=require_writable)
+    dirs = result["directories"]
+    return {**result, "path": dirs[0] if dirs else ""}
+
+
+# The per-app wrappers MNP still names. They exist so an MNP 1.0 client's
+# `video_root` / `audio_root` / `photo_roots` messages keep working; nothing
+# new should be added here — a new app calls the generic pair above.
+
+async def set_video_root(state: dict, group_id: str, path: str) -> dict:
+    result = await set_app_directory(state, group_id, "video", path)
+    return {"path": result["path"], "group_id": group_id}
+
+
+async def set_audio_root(state: dict, group_id: str, path: str) -> dict:
+    # "music", not "audio": the app's registry key is what identifies it
+    # everywhere, and `audio_root` is only the name the setting used to have.
+    result = await set_app_directory(state, group_id, "music", path)
+    return {"path": result["path"], "group_id": group_id}
 
 
 async def set_photo_roots(state: dict, group_id: str, roots: list[str]) -> dict:
+    result = await set_app_directories(state, group_id, "photo", roots)
+    return {"roots": result["directories"], "group_id": group_id}
+
+
+# ── Chat ─────────────────────────────────────────────────────────────────────
+
+async def set_chat_directory(state: dict, group_id: str, path: str) -> dict:
+    """
+    Where chat attachments are written.
+
+    `require_writable`, unlike every other app directory: this one is a
+    *destination*, not a view. Pointing it at a read-only root would produce an
+    attachment button that fails at the moment somebody uses it, which is the
+    failure mode the RO/RW model exists to move earlier.
     """
-    Which folder(s) are the Photos app's entry points for this group. Unlike
-    `set_video_root`/`set_audio_root`, the whole *set* is replaced in one
-    call (docs/photos.md §2.1) — signed once, same shape as
-    `set_enabled_apps`, rather than one op per root added/removed.
+    return await set_app_directory(state, group_id, "chat", path,
+                                   require_writable=True)
 
-    Always fires a sweep, even to an empty list: a root just added needs its
-    existing contents enriched (nothing else re-visits already-indexed
-    entries), and a root just removed leaves its cache entries harmlessly
-    unused rather than needing any cleanup — re-sweeping the new set costs
-    nothing when it's empty.
+
+async def set_chat_link_preview(state: dict, group_id: str,
+                                enabled: bool) -> dict:
+    """
+    Whether the node fetches a page's title and image when a member posts a
+    link.
+
+    Outbound third-party traffic on the operator's connection, caused by a
+    message they did not write and pointing at a URL they did not choose — so
+    it is theirs to switch off, on the same reasoning as the per-group TMDB
+    switch. Absent means on, because that is what the node did before this
+    existed.
     """
     roster = _roster(state)
     ctx = _group_ctx(state, group_id)
-    await roster.set_photo_roots(group_id, roots, set_by=state.get("node_user_id", ""))
-    ctx["photo_roots"] = roots
-    log.info("Photo roots for group %s: %s", group_id[:8], ", ".join(sorted(roots)) or "(none)")
-    enrich_fn = state.get("enrich_photo_roots_fn")
-    if enrich_fn:
-        asyncio.ensure_future(enrich_fn(group_id))
-    return {"roots": roots, "group_id": group_id}
+    await roster.set_chat_link_preview(group_id, enabled,
+                                       set_by=state.get("node_user_id", ""))
+    ctx["chat_link_preview"] = enabled
+    log.info("Chat link previews for group %s: %s", group_id[:8],
+             "on" if enabled else "off")
+    return {"enabled": enabled, "group_id": group_id}
 
 
 # ── Scan settings ────────────────────────────────────────────────────────────
@@ -1086,7 +1397,7 @@ async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs:
     """
     How often the indexer's reconciliation backstop runs, and how long a
     changed file is left alone before being hashed (indexer.py
-    DirectoryIndexer). Persisted like set_member_upload/set_enabled_apps —
+    DirectoryIndexer). Persisted like set_enabled_apps —
     but there is also a *live* DirectoryIndexer object to update, since it
     reads these once at construction and runs its own background loop with
     them rather than consulting groups_ctx on every use.
diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py
index 74ea2f6..d288231 100644
--- a/packages/meshbay-node/src/meshbay_node/roots.py
+++ b/packages/meshbay-node/src/meshbay_node/roots.py
@@ -120,10 +120,9 @@ class Root:
     name: str
     path: Path
     kind: str = "generic"
-    upload: bool = False
-    direct: bool = False
-    # Runtime, not configuration: set by the indexer when the directory can no
-    # longer be read, and cleared when it comes back.
+    writable: bool = False
+    removable: bool = False
+    ejected: bool = False
     available: bool = True
 
     @property
@@ -165,6 +164,13 @@ class RootSet:
 
     roots: list[Root] = field(default_factory=list)
 
+    # Roots this set ejected by itself — a removable device that went away
+    # without the operator clicking Eject. Drained by the indexer, which is
+    # the only caller holding a roster to write the state to. Without that
+    # the flag is lost on the next restart, and the surprise unplug looks
+    # like a deletion all over again on the pass after it.
+    auto_ejected: list[str] = field(default_factory=list)
+
     # ── Construction ─────────────────────────────────────────────────────────
 
     @classmethod
@@ -172,8 +178,9 @@ class RootSet:
         """
         Build from configuration, refusing anything ambiguous.
 
-        `specs` are dicts with `path`, and optionally `name`, `kind`, `upload`.
-        Raises RootError with a message meant for an operator reading a log.
+        `specs` are dicts with `path`, and optionally `name`, `kind`, `writable`,
+        `removable`. Raises RootError with a message meant for an operator reading
+        a log.
         """
         roots: list[Root] = []
         by_folded: dict[str, Root] = {}
@@ -209,34 +216,22 @@ class RootSet:
                 log.warning("root %r: unknown kind %r — using 'generic'", name, kind)
                 kind = "generic"
 
+            # Backward compat: old configs use `upload` instead of `writable`
+            writable = bool(spec.get("writable", spec.get("upload", False)))
+            # `ejected` is runtime state, not configuration — it reaches here
+            # only from the roster, restored at startup so a drive ejected
+            # before a restart does not come back on its own.
             root = Root(name=name, path=path, kind=kind,
-                        upload=bool(spec.get("upload", False)),
-                        direct=bool(spec.get("direct", False)))
+                        writable=writable,
+                        removable=bool(spec.get("removable", False)),
+                        ejected=bool(spec.get("ejected", False)),
+                        available=not bool(spec.get("ejected", False)))
             _refuse_nesting(root, roots)
             roots.append(root)
             by_folded[root.folded] = root
 
-        cls._settle_upload_root(roots)
         return cls(roots=roots)
 
-    @staticmethod
-    def _settle_upload_root(roots: list[Root]) -> None:
-        """
-        Exactly one root receives uploads, and the operator picks it.
-
-        Not guessed when several are marked, because "uploads went somewhere
-        else" is discovered weeks later. With none marked and a single root, the
-        answer is not ambiguous, so it is taken.
-        """
-        marked = [r for r in roots if r.upload]
-        if len(marked) > 1:
-            names = ", ".join(r.name for r in marked)
-            raise RootError(
-                f"several roots are marked upload = true ({names}) — exactly one "
-                f"receives uploads")
-        if not marked and len(roots) == 1:
-            roots[0].upload = True
-
     # ── Lookup ───────────────────────────────────────────────────────────────
 
     def by_name(self, name: str) -> Root | None:
@@ -247,11 +242,8 @@ class RootSet:
         return None
 
     @property
-    def upload_root(self) -> Root | None:
-        for root in self.roots:
-            if root.upload:
-                return root
-        return None
+    def writable_roots(self) -> list[Root]:
+        return [r for r in self.roots if r.writable]
 
     @property
     def names(self) -> list[str]:
@@ -336,10 +328,28 @@ class RootSet:
         Called periodically and after a filesystem event that looks like a
         disappearance. A change here never edits the index: a root going away
         freezes its entries, and a root coming back triggers a rescan.
+
+        An ejected root stays unavailable regardless of `is_live()` — the
+        operator must explicitly plug it back. A removable root whose path
+        disappears without an eject is auto-ejected as a safety net.
         """
         changed: list[tuple[Root, bool]] = []
         for root in self.roots:
+            if root.ejected:
+                if root.available:
+                    root.available = False
+                    changed.append((root, False))
+                continue
             live = root.is_live()
+            if not live and root.removable:
+                root.ejected = True
+                # Recorded for the caller to persist. A flag that only lives
+                # in memory would be forgotten on the next restart, and the
+                # rescan that followed would read an empty mount point as an
+                # erased library — the exact outcome eject exists to prevent.
+                self.auto_ejected.append(root.name)
+                log.warning("Root %r auto-ejected (device disappeared): %s",
+                            root.name, root.path)
             if live != root.available:
                 root.available = live
                 changed.append((root, live))
@@ -347,14 +357,31 @@ class RootSet:
                             "available" if live else "UNAVAILABLE", root.path)
         return changed
 
-    def describe(self) -> list[dict]:
-        """Per-root state for the index payload and the admin UI."""
+    def describe(self, *, with_paths: bool = False) -> list[dict]:
+        """
+        Per-root state for the index payload and the admin UI.
+
+        Deliberately no paths by default: this is what every member receives.
+        `with_paths=True` is the operator's own view, over a channel that is
+        already theirs alone (loopback + run token).
+        """
         out = []
         for r in self.roots:
             d: dict = {"name": r.name, "kind": r.kind,
-                        "available": r.available, "upload": r.upload}
-            if r.direct:
-                d["direct"] = True
+                        "available": r.available,
+                        "writable": r.writable,
+                        "removable": r.removable,
+                        "ejected": r.ejected,
+                        # Backward compat for MNP 1.0 clients
+                        "upload": r.writable}
+            # `with_paths` is for the operator's *own* channels only — the
+            # loopback API and the CLI reading it, both of which already
+            # require being on this machine with the run token. A member is
+            # told what exists and whether it is readable, never where on the
+            # operator's disk it lives, and the index payload every member
+            # receives must keep calling this without the flag.
+            if with_paths:
+                d["path"] = str(r.path)
             out.append(d)
         return out
 
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index c78281b..af87f92 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -31,6 +31,8 @@ from pathlib import Path
 
 import aiosqlite
 
+from meshbay_common.paths import fold
+
 log = logging.getLogger(__name__)
 
 # Crockford base32 without I, L, O and U: no character pair a human can confuse
@@ -543,10 +545,40 @@ class Roster:
 
     # ── Group settings ──────────────────────────────────────────────────────
 
-    # Whether members who are not the operator may upload. Default is yes: a
-    # group that nobody may add to is the unusual case, and an existing node
-    # must not change behaviour because a table was added under it.
-    SETTING_MEMBER_UPLOAD = "member_upload"
+    # Whether a root is ejected. Runtime state, one key per root, keyed by the
+    # *folded* name so it agrees with the case-insensitive comparison the rest
+    # of the root code makes. It lives here rather than in node.toml because it
+    # is not configuration — an operator's hand-written config file should not
+    # be rewritten because a USB drive was unplugged — and it has to survive a
+    # restart, or the rescan that follows reads an empty mount point as an
+    # erased library, which is the whole thing eject exists to prevent.
+    SETTING_ROOT_EJECTED_PREFIX = "root_ejected:"
+
+    @classmethod
+    def root_ejected_key(cls, root_name: str) -> str:
+        return cls.SETTING_ROOT_EJECTED_PREFIX + fold(root_name)
+
+    async def set_root_ejected(self, group_id: str, root_name: str,
+                               ejected: bool, set_by: str = "") -> None:
+        await self.set_setting(group_id, self.root_ejected_key(root_name),
+                               "1" if ejected else "0", set_by)
+
+    async def ejected_roots(self, group_id: str) -> set[str]:
+        """
+        The folded names of this group's ejected roots.
+
+        Matched in Python rather than with `LIKE 'root_ejected:%'`: `_` is a
+        single-character wildcard there, so that pattern also matches keys this
+        does not own. A group has a handful of settings rows, so reading them
+        all costs nothing and the prefix test is then exact.
+        """
+        prefix = self.SETTING_ROOT_EJECTED_PREFIX
+        async with self._db.execute(
+                "SELECT key, value FROM group_settings WHERE group_id = ?",
+                (group_id,)) as cur:
+            rows = await cur.fetchall()
+        return {r["key"][len(prefix):] for r in rows
+                if r["key"].startswith(prefix) and r["value"] == "1"}
 
     async def get_setting(self, group_id: str, key: str,
                           default: str | None = None) -> str | None:
@@ -567,17 +599,6 @@ class Roster:
             (group_id, key, value, set_by, _now()))
         await self._db.commit()
 
-    async def member_upload_allowed(self, group_id: str) -> bool:
-        """Whether an ordinary member may upload to this group."""
-        value = await self.get_setting(group_id, self.SETTING_MEMBER_UPLOAD, "1")
-        return value != "0"
-
-    async def set_member_upload(self, group_id: str, allowed: bool,
-                                set_by: str = "") -> bool:
-        await self.set_setting(group_id, self.SETTING_MEMBER_UPLOAD,
-                               "1" if allowed else "0", set_by)
-        return allowed
-
     # Which group "applications" (Chat, Files, and whatever registers later in
     # apps.js) are shown to members. Unset means every app that exists — an
     # existing group's tabs must not disappear because a node was upgraded.
@@ -587,11 +608,15 @@ class Roster:
     async def enabled_apps(self, group_id: str) -> list[str]:
         value = await self.get_setting(group_id, self.SETTING_ENABLED_APPS)
         if value is None:
-            return list(self.DEFAULT_APPS)
-        try:
-            return list(json.loads(value))
-        except (ValueError, TypeError):
-            return list(self.DEFAULT_APPS)
+            apps = list(self.DEFAULT_APPS)
+        else:
+            try:
+                apps = list(json.loads(value))
+            except (ValueError, TypeError):
+                apps = list(self.DEFAULT_APPS)
+        if "files" not in apps:
+            apps.insert(0, "files")
+        return apps
 
     async def set_enabled_apps(self, group_id: str, apps: list[str],
                                set_by: str = "") -> list[str]:
@@ -605,7 +630,7 @@ class Roster:
     # user_id)` authorizing the operator node-wide (desktop-client-v1.md
     # §6.3). Unset means "the shipped default token, TMDB's own default
     # language" — the same "absent means the old behaviour" discipline
-    # member_upload/enabled_apps already follow.
+    # enabled_apps already follows.
     #
     # Whether TMDB is used *at all*, though, is per-group (moved off the
     # node-wide sentinel below, 2026-08-24): an operator running a real media
@@ -636,58 +661,110 @@ class Roster:
             await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE,
                                    language, set_by)
 
-    # Which folder is the Videos app's entry point for this group — per-group
-    # (unlike the token/language above), since different groups share
-    # different trees. Empty/unset means the whole group index, exactly as
-    # today.
-    SETTING_VIDEO_ROOT = "video_root"
-
-    async def video_root(self, group_id: str) -> str:
-        return await self.get_setting(group_id, self.SETTING_VIDEO_ROOT, "") or ""
+    # ── App directories ─────────────────────────────────────────────────────
+    #
+    # Which folder(s) inside the group's shared roots each application uses as
+    # its entry point. One storage shape for every app, keyed by the app's own
+    # name, so adding an application needs no change here at all — that is the
+    # whole point of the plugin architecture (docs/refactor-groups.md §1.6).
+    #
+    # Always a JSON list, even for an app that only ever wants one directory.
+    # Two shapes for one idea is how `video_root` (scalar) and `photo_roots`
+    # (list) ended up needing separate ops, separate MNP messages and separate
+    # settings widgets to say the same thing.
+    #
+    # Empty/unset means nothing configured yet, and every app reads that as
+    # "show nothing until an operator has chosen" rather than "the whole group
+    # index" — the discipline video_root established, kept.
+    SETTING_APP_DIRS_SUFFIX = "_directories"
 
-    async def set_video_root(self, group_id: str, path: str, set_by: str = "") -> str:
-        await self.set_setting(group_id, self.SETTING_VIDEO_ROOT, path or "", set_by)
-        return path or ""
+    # What each app's directories used to be stored under, before they were
+    # one shape. Read as a fallback so an existing node keeps working with no
+    # migration step: the legacy key is never written again, and the first
+    # save through the new path leaves it behind.
+    # Keyed by the *registry* name the app is known by everywhere else
+    # (apps.js, ALLOWED_APPS, enabled_apps) — which for Music is "music", while
+    # its old setting was called `audio_root`. One identifier per app, and the
+    # place the two names meet is this table and nowhere else.
+    LEGACY_DIR_KEYS = {
+        "video": ("video_root", "scalar"),
+        "music": ("audio_root", "scalar"),
+        "photo": ("photo_roots", "list"),
+    }
 
-    # 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"
+    # The name each app's directories are *also* published under, for readers
+    # that predate the list — the handshake ack's `video_root`, and the group
+    # context the ack builds from. Derived from the list, never stored beside
+    # it, so the two cannot disagree; the shape says how to derive it.
+    CTX_ALIASES = {
+        "video": ("video_root", "scalar"),
+        "music": ("audio_root", "scalar"),
+        "photo": ("photo_roots", "list"),
+        "chat": ("chat_directory", "scalar"),
+    }
 
-    async def audio_root(self, group_id: str) -> str:
-        return await self.get_setting(group_id, self.SETTING_AUDIO_ROOT, "") or ""
+    @classmethod
+    def app_dirs_key(cls, app_key: str) -> str:
+        return f"{app_key}{cls.SETTING_APP_DIRS_SUFFIX}"
 
-    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 ""
+    @classmethod
+    def ctx_alias(cls, app_key: str, directories: list[str]) -> tuple[str, object] | None:
+        """The (name, value) an app's directories are also published under."""
+        alias = cls.CTX_ALIASES.get(app_key)
+        if not alias:
+            return None
+        name, shape = alias
+        if shape == "list":
+            return name, list(directories)
+        return name, (directories[0] if directories else "")
 
-    # Which folder(s) are the Photos app's entry points for this group —
-    # a *set*, unlike video_root/audio_root above: a photo library is
-    # routinely scattered across several unrelated folders (docs/photos.md
-    # §2.1), so there is no single natural root to pick. Stored the same way
-    # `enabled_apps` already is (json.dumps(sorted(...))). Empty/unset means
-    # nothing configured yet — same "show nothing until an operator has
-    # chosen" discipline video_root/audio_root already established, not
-    # "the whole group index".
-    SETTING_PHOTO_ROOTS = "photo_roots"
+    async def app_directories(self, group_id: str, app_key: str) -> list[str]:
+        value = await self.get_setting(group_id, self.app_dirs_key(app_key))
+        if value is not None:
+            try:
+                return [str(p) for p in json.loads(value)]
+            except (ValueError, TypeError):
+                return []
 
-    async def photo_roots(self, group_id: str) -> list[str]:
-        value = await self.get_setting(group_id, self.SETTING_PHOTO_ROOTS)
-        if value is None:
+        legacy = self.LEGACY_DIR_KEYS.get(app_key)
+        if not legacy:
             return []
+        key, shape = legacy
+        raw = await self.get_setting(group_id, key)
+        if raw is None:
+            return []
+        if shape == "scalar":
+            return [raw] if raw else []
         try:
-            return list(json.loads(value))
+            return [str(p) for p in json.loads(raw)]
         except (ValueError, TypeError):
             return []
 
-    async def set_photo_roots(self, group_id: str, roots: list[str],
-                              set_by: str = "") -> list[str]:
-        await self.set_setting(group_id, self.SETTING_PHOTO_ROOTS,
-                               json.dumps(sorted(roots)), set_by)
-        return roots
+    async def set_app_directories(self, group_id: str, app_key: str,
+                                  paths: list[str], set_by: str = "") -> list[str]:
+        clean = sorted({str(p).strip("/") for p in paths if str(p).strip("/")})
+        await self.set_setting(group_id, self.app_dirs_key(app_key),
+                               json.dumps(clean), set_by)
+        return clean
+
+    # ── Chat ────────────────────────────────────────────────────────────────
+
+    # Whether the node fetches a page's title/preview when a member posts a
+    # link. Outbound third-party traffic on the operator's connection, from a
+    # message they did not write, so it is theirs to switch off — the same
+    # reasoning as the per-group TMDB switch. Unset means on, because that is
+    # what the node did before this existed.
+    SETTING_CHAT_LINK_PREVIEW = "chat_link_preview"
+
+    async def chat_link_preview(self, group_id: str) -> bool:
+        value = await self.get_setting(group_id, self.SETTING_CHAT_LINK_PREVIEW, "1")
+        return value != "0"
+
+    async def set_chat_link_preview(self, group_id: str, enabled: bool,
+                                    set_by: str = "") -> bool:
+        await self.set_setting(group_id, self.SETTING_CHAT_LINK_PREVIEW,
+                               "1" if enabled else "0", set_by)
+        return enabled
 
     # 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
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 94dfd8e..8e357c9 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -76,8 +76,14 @@ from meshbay_common.adminop import (
     OP_MUSICBRAINZ_ENABLED,
     OP_AUDIO_ROOT,
     OP_PHOTO_ROOTS,
+    OP_APP_DIRECTORIES,
+    OP_CHAT_DIRECTORY,
+    OP_CHAT_LINK_PREVIEW,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
+    OP_ROOT_UPDATE,
+    OP_ROOT_EJECT,
+    OP_ROOT_PLUG,
     OP_GROUP_ATTACH,
     OP_GROUP_DETACH,
     admin_transcript,
@@ -208,7 +214,6 @@ JOIN_FAILURE_WINDOW = 600  # seconds
 # attachments from the chat alike. One visible directory the operator can look
 # into, back up or empty — rather than a hidden tree of per-user uuids that
 # nobody could read, or files scattered wherever someone happened to be looking.
-UPLOAD_DIR_NAME = "uploads"
 
 
 def _extract_dtls_fingerprint(sdp: str) -> bytes:
@@ -481,6 +486,12 @@ class WebRTCPeerSession:
                 self._do_audio_root(msg)
             elif mtype == MNP.PHOTO_ROOTS:
                 self._do_photo_roots(msg)
+            elif mtype == MNP.APP_DIRECTORIES:
+                self._do_app_directories(msg)
+            elif mtype == MNP.CHAT_DIRECTORY:
+                self._do_chat_directory(msg)
+            elif mtype == MNP.CHAT_LINK_PREVIEW:
+                self._do_chat_link_preview(msg)
             elif mtype == MNP.MEDIA_META_REQ:
                 self._spawn(self._do_media_meta_request(msg))
             elif mtype == MNP.SEASON_META_REQ:
@@ -507,6 +518,12 @@ class WebRTCPeerSession:
                 self._do_root_add(msg)
             elif mtype == MNP.ROOT_REMOVE:
                 self._do_root_remove(msg)
+            elif mtype == MNP.ROOT_UPDATE:
+                self._do_root_update(msg)
+            elif mtype == MNP.ROOT_EJECT:
+                self._do_root_eject(msg)
+            elif mtype == MNP.ROOT_PLUG:
+                self._do_root_plug(msg)
             elif mtype == MNP.ROSTER_READ:
                 self._spawn(self._do_roster_read(msg))
             elif mtype == MNP.DENYLIST_READ:
@@ -754,10 +771,12 @@ class WebRTCPeerSession:
         # channel and nothing else.
         config = {
             "is_node_admin": self._is_node_admin(),
-            # So the interface knows whether to offer uploading at all. Not a
-            # permission — the node refuses regardless — but without it the
-            # only way to discover the answer is to try.
-            "member_upload": bool(self._group_ctx().get("member_upload", True)),
+            # Backward compat for MNP 1.0 clients: computed from writable roots.
+            # New clients read per-root writable from the index payload instead.
+            "member_upload": any(
+                r.get("writable") for r in
+                (self._group_ctx().get("roots").describe()
+                 if self._group_ctx().get("roots") else [])),
             # Which group "applications" to show. Absent/empty falls back to
             # every registered one client-side, so a node that predates this
             # setting (or one whose context has not loaded it yet) hides
@@ -791,6 +810,23 @@ class WebRTCPeerSession:
             # (docs/photos.md §2.1). Empty means unset (the Photos tab shows
             # nothing yet).
             "photo_roots": list(self._group_ctx().get("photo_roots") or []),
+            # The same three answers in one shape, plus every other app's —
+            # `<app>_directories`, keyed by the app's registry name, always a
+            # list. The scalars above are derived from these (daemon.py's
+            # `_app_directories_ctx`) and kept for MNP 1.0 clients, which can
+            # represent one folder and never more. A client reading the plural
+            # form gets all of them.
+            **{f"{app}_directories":
+               list(self._group_ctx().get(f"{app}_directories") or [])
+               for app in ("video", "music", "photo", "chat")},
+            # Where chat attachments are written — the singular form, because
+            # Chat genuinely has one destination. "" means the operator has not
+            # chosen, and the paperclip says so.
+            "chat_directory": self._group_ctx().get("chat_directory") or "",
+            # Whether the node unfurls links members post here. Absent means
+            # on, which is what it did before this existed.
+            "chat_link_preview": bool(
+                self._group_ctx().get("chat_link_preview", True)),
             # 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
@@ -1559,6 +1595,28 @@ class WebRTCPeerSession:
                         "detail": "Choose a folder to create this in"})
             return
 
+        # Read-only means read-only, and creating a folder writes to the
+        # operator's disk. `_do_file_upload` gained this check with the RO/RW
+        # model and this one did not — so a member could not add a file to a
+        # published library but could still leave empty directories in it.
+        owner = roots.split(parent_rel)
+        if owner is None:
+            self._send({"type": "error", "detail": "Invalid directory"})
+            return
+        parent_root, _tail = owner
+        if not parent_root.writable:
+            self._send({"type": "error",
+                        "detail": f"Directory '{parent_root.name}' is read-only",
+                        "code": "root_read_only"})
+            self._audit("dir_create_refused", parent_rel[:64])
+            return
+        if not parent_root.available:
+            self._send({"type": "error",
+                        "detail": f"Directory '{parent_root.name}' is "
+                                  f"currently unavailable",
+                        "code": "root_unavailable"})
+            return
+
         parent = safe_subdir(roots, parent_rel)
         if parent is None or not parent.is_dir():
             self._send({"type": "error", "detail": "Invalid directory"})
@@ -1756,52 +1814,12 @@ class WebRTCPeerSession:
                     "user_id": user_id})
 
     def _do_member_upload(self, msg: dict) -> None:
-        """
-        Turn uploading by ordinary members on or off, for this group.
-
-        Signed like every other operator action. The setting decides who may
-        write to the operator's disk, so a node that took it from an unsigned
-        message would let any member turn it back on for everyone — the control
-        would be a suggestion.
-        """
-        if "allowed" not in msg:
-            self._send({"type": "error", "detail": "Missing allowed"})
-            return
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for this"})
-            return
-        # The subject is what the operator is shown before signing, so it has to
-        # name the outcome rather than the operation.
-        self._issue_admin_challenge(
-            OP_MEMBER_UPLOAD, "on" if msg.get("allowed") else "off")
-
-    async def _admin_exec_member_upload(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        allowed = pending["subject"] == "on"
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"member_upload:{pending['subject']}")
-            return
-        try:
-            await self._run_op(
-                ops.set_member_upload, self._group_id or "", allowed)
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("member_upload", pending["subject"])
-
-        # Everyone already connected is told, rather than finding out by having
-        # an upload refused. Enforcement does not depend on this reaching them —
-        # it is the node that refuses — but a button that stays visible until
-        # the next reconnection is a button people press.
-        notice = {"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
-                  "allowed": allowed}
-        for uid, session in list(self._peer_registry().items()):
-            try:
-                session._send(notice)
-            except Exception:
-                pass
+        # Deprecated: upload control is now per-root via writable flag.
+        # Old clients may still send this — acknowledge without acting.
+        log.warning("Deprecated member_upload message received — use root "
+                    "writable/read-only instead")
+        self._send({"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
+                    "allowed": True, "deprecated": True})
 
     # Every "application" a group can show. Photos joins this set (and
     # apps.js's registry, client-side) when it lands; nothing else about
@@ -1810,13 +1828,20 @@ class WebRTCPeerSession:
     # network calls (TMDB, MusicBrainz) once enabled, so an operator opts a
     # group in explicitly rather than getting it for free
     # (docs/mediacenter.md §5.6, docs/musicbay.md §4.4).
-    ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo"})
+    # `helloworld` is the reference implementation (docs/refactor-groups.md
+    # §4.1), hidden client-side behind `?dev=1`. It is here because the
+    # allow-list is server-side enforcement — a client that names an app this
+    # node does not know is refused — and an app the node refused could not
+    # demonstrate anything. This entry and the client's registry line are the
+    # whole of what adding an application costs.
+    ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo",
+                              "helloworld"})
 
     def _do_apps_enabled(self, msg: dict) -> None:
         """
         Turn a group "application" on or off for everyone, for this group.
 
-        Signed like `member_upload`: this decides what a member sees, and an
+        Signed like the root ops: this decides what a member sees, and an
         unsigned message would let any member turn a disabled one back on.
         """
         apps = msg.get("apps")
@@ -1828,6 +1853,12 @@ class WebRTCPeerSession:
             self._send({"type": "error",
                        "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"})
             return
+        # Files is not a toggle: MNP permits root exploration regardless of
+        # what this list says, so hiding the tab only ever misled. Added at the
+        # front, the same order ops.set_enabled_apps writes, so the landing-tab
+        # preference sees one list and not two.
+        if "files" not in apps:
+            apps.insert(0, "files")
         if not self._has_admin_authority():
             self._send({"type": "error", "detail": "No authorized key for this"})
             return
@@ -1911,7 +1942,7 @@ class WebRTCPeerSession:
         self._audit("tmdb_config", pending["subject"])
 
         # Node-wide setting: every connected peer in every group is told, not
-        # just this group's peers (unlike apps_enabled/member_upload/the
+        # just this group's peers (unlike apps_enabled/the root ops/the
         # per-group tmdb_enabled below).
         notice = {
             "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION,
@@ -2119,6 +2150,150 @@ class WebRTCPeerSession:
             except Exception:
                 pass
 
+    # ── App directories (generic) ────────────────────────────────────────
+
+    def _do_app_directories(self, msg: dict) -> None:
+        """
+        Which folder(s) an application works over, for any application.
+
+        One handler where there were three near-identical ones (`video_root`,
+        `audio_root`, `photo_roots`) differing only in the key they wrote and
+        whether they carried a string or a list. Those three still exist for
+        clients that speak them; nothing new is added beside them.
+
+        `app` must be one this node knows (`ALLOWED_APPS`) — a client-supplied
+        key is otherwise a way to write arbitrary rows into `group_settings`.
+        The paths are checked by `ops._validate_app_dirs`, which runs after the
+        signature: this is a settings change, not a capability, so refusing
+        early here would be a courtesy rather than the control.
+        """
+        app = str(msg.get("app", "")).strip()
+        dirs = msg.get("directories")
+        if app not in self.ALLOWED_APPS:
+            self._send({"type": "error", "detail": f"Unknown app {app!r}"})
+            return
+        if not isinstance(dirs, list) or not all(isinstance(d, str) for d in dirs):
+            self._send({"type": "error",
+                        "detail": "Missing or invalid 'directories'"})
+            return
+        clean = sorted({d.strip("/") for d in dirs if d.strip("/")})
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        # The app is in the subject, not only the paths: an operator shown
+        # "Media/Films" alone cannot tell which application is about to be
+        # pointed at it, and two apps' challenges would be indistinguishable.
+        self._issue_admin_challenge(
+            OP_APP_DIRECTORIES, f"{app}:{','.join(clean)}")
+
+    async def _admin_exec_app_directories(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        app, _, joined = pending["subject"].partition(":")
+        dirs = joined.split(",") if joined else []
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"app_directories:{pending['subject']}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_app_directories, self._group_id or "", app, dirs)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("app_directories", pending["subject"])
+        self._broadcast_to_group({"type": MNP.APP_DIRECTORIES_ACK,
+                                  "v": MNP_VERSION, "app": app,
+                                  "directories": result["directories"]})
+
+    # ── Chat ─────────────────────────────────────────────────────────────
+
+    def _do_chat_directory(self, msg: dict) -> None:
+        """
+        Where chat attachments are written.
+
+        Unlike every other app directory this one is a destination, so it has
+        to be on a read-write root — checked by `ops.set_chat_directory` after
+        the signature, which is where the refusal actually lives.
+        """
+        path = msg.get("path")
+        if not isinstance(path, str):
+            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
+            return
+        path = path.strip("/")
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_CHAT_DIRECTORY, path)
+
+    async def _admin_exec_chat_directory(
+        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"chat_directory:{path}")
+            return
+        try:
+            await self._run_op(
+                ops.set_chat_directory, self._group_id or "", path)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_directory", path)
+        self._broadcast_to_group(
+            {"type": MNP.CHAT_DIRECTORY_ACK, "v": MNP_VERSION, "path": path})
+
+    def _do_chat_link_preview(self, msg: dict) -> None:
+        """
+        Whether the node fetches a page's title and image when a member posts
+        a link — outbound traffic on the operator's connection, from a message
+        they did not write, so it is signed like everything else that decides
+        what leaves this machine.
+        """
+        enabled = msg.get("enabled")
+        if not isinstance(enabled, bool):
+            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_CHAT_LINK_PREVIEW, "on" if enabled else "off")
+
+    async def _admin_exec_chat_link_preview(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        enabled = pending["subject"] == "on"
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"chat_link_preview:{pending['subject']}")
+            return
+        try:
+            await self._run_op(
+                ops.set_chat_link_preview, self._group_id or "", enabled)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_link_preview", pending["subject"])
+        self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK,
+                                  "v": MNP_VERSION, "enabled": enabled})
+
+    def _broadcast_to_group(self, notice: dict) -> None:
+        """
+        Tell everyone connected to this group about a setting that changed.
+
+        Enforcement never depends on this reaching them — the node is what
+        refuses — but a control that stays on screen until the next
+        reconnection is a control people use.
+        """
+        for _uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     def _do_musicbrainz_enabled(self, msg: dict) -> None:
         """
         Whether MusicBrainz lookups run for this group at all. Per-group
@@ -2308,11 +2483,14 @@ class WebRTCPeerSession:
         if not self._has_admin_authority():
             self._send({"type": "error", "detail": "No authorized key for this"})
             return
-        upload_dir = str(msg.get("upload_dir", "")).strip()
+        # `upload_dir` is not read here any more, and a client still sending it
+        # is ignored rather than obeyed: on load it forces every other root
+        # read-only, which is the model the RO/RW one replaced. A second
+        # writable directory is `root_add` with `writable`.
         self._issue_admin_challenge(
             OP_GROUP_ATTACH, name,
             payload={"name": name, "shared_dir": shared_dir,
-                     "upload_dir": upload_dir},
+                     "writable": bool(msg.get("writable", True))},
             group_id="")
 
     async def _admin_exec_group_attach(
@@ -2326,7 +2504,8 @@ class WebRTCPeerSession:
         p = pending.get("payload") or {}
         try:
             result = await self._run_op(
-                ops.attach_group, p["name"], p["shared_dir"], p.get("upload_dir", ""))
+                ops.attach_group, p["name"], p["shared_dir"],
+                writable=bool(p.get("writable", True)))
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
@@ -2409,7 +2588,8 @@ class WebRTCPeerSession:
                 "group_id": target_group, "path": path,
                 "name": str(msg.get("name", ""))[:128],
                 "kind": str(msg.get("kind", "generic"))[:16],
-                "upload": bool(msg.get("upload", False)),
+                "writable": bool(msg.get("writable", msg.get("upload", False))),
+                "removable": bool(msg.get("removable", False)),
             },
             group_id=target_group)
 
@@ -2425,7 +2605,8 @@ class WebRTCPeerSession:
             result = await self._run_op(
                 ops.add_root, p["group_id"], p["path"],
                 name=p.get("name", ""), kind=p.get("kind", "generic"),
-                upload=p.get("upload", False))
+                writable=p.get("writable", False),
+                removable=p.get("removable", False))
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
@@ -2473,6 +2654,141 @@ class WebRTCPeerSession:
         await self._retarget_indexer(p["group_id"])
         self._send({"type": MNP.ROOT_REMOVE_ACK, "v": MNP_VERSION, **result})
 
+    def _do_root_update(self, msg: dict) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        updates = []
+        if "writable" in msg:
+            updates.append(f"rw={'on' if msg['writable'] else 'off'}")
+        if "removable" in msg:
+            updates.append(f"rem={'on' if msg['removable'] else 'off'}")
+        subject = f"{root_name}:{','.join(updates)}" if updates else root_name
+        self._issue_admin_challenge(
+            OP_ROOT_UPDATE, subject,
+            payload={
+                "group_id": target_group, "root_name": root_name,
+                "writable": msg.get("writable"),
+                "removable": msg.get("removable"),
+            },
+            group_id=target_group)
+
+    async def _admin_exec_root_update(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"root_update:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(
+                ops.update_root, p["group_id"], p["root_name"],
+                writable=p.get("writable"), removable=p.get("removable"))
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("root_update failed: %s", e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit("root_update", pending["subject"])
+        await self._retarget_indexer(p["group_id"])
+        notice = {"type": MNP.ROOT_UPDATE_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    def _do_root_eject(self, msg: dict) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_ROOT_EJECT, root_name,
+            payload={"group_id": target_group, "root_name": root_name},
+            group_id=target_group)
+
+    async def _admin_exec_root_eject(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"root_eject:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(
+                ops.eject_root, p["group_id"], p["root_name"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("root_eject failed: %s", e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit("root_eject", p["root_name"])
+        notice = {"type": MNP.ROOT_EJECT_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    def _do_root_plug(self, msg: dict) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_ROOT_PLUG, root_name,
+            payload={"group_id": target_group, "root_name": root_name},
+            group_id=target_group)
+
+    async def _admin_exec_root_plug(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"root_plug:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(
+                ops.plug_root, p["group_id"], p["root_name"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("root_plug failed: %s", e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit("root_plug", p["root_name"])
+        notice = {"type": MNP.ROOT_PLUG_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     async def _run_op(self, fn, *args, **kwargs):
         """
         Call an operation from `meshbay_node.ops` with the daemon's own view.
@@ -2489,10 +2805,36 @@ class WebRTCPeerSession:
         return await fn(state, *args, **kwargs)
 
     async def _retarget_indexer(self, group_id: str) -> None:
-        """Tell the indexer to rescan after roots changed."""
+        """
+        Pick up a root that was just added to or removed from node.toml.
+
+        Through the daemon's own reload, which is what the loopback API has
+        always done after the same operations (`ui/app.py`). This used to
+        re-point the indexer at `groups_ctx[gid]["roots"]` instead — the very
+        object the op had just edited — so `retarget` diffed a set against
+        itself, found no new names, scanned nothing, and dropped nothing. A
+        directory added over MNP reached node.toml and was invisible until a
+        restart; one removed kept serving its files.
+
+        Two front doors doing different things is the shape `ops.py` exists to
+        prevent, and this was it: the loopback path worked and the MNP path did
+        not, which is why it survived until the operator added a directory from
+        a browser.
+
+        Not awaited: a reload rescans, and a new library is minutes. The ack
+        the caller sends carries the set the node is moving to, and the
+        `index_sync` that follows the scan carries what it found.
+        """
         state = self._ctx.get("daemon_state")
         if not state:
             return
+        reload_fn = state.get("reload_fn")
+        if reload_fn:
+            self._spawn(reload_fn())
+            return
+        # No daemon to ask — a test harness, or a context assembled by hand.
+        # Retarget directly, which is correct as long as the caller did not
+        # edit the live set in place.
         indexer = state.get("indexers", {}).get(group_id)
         roots = state.get("groups_ctx", {}).get(group_id, {}).get("roots")
         if indexer and roots:
@@ -3620,6 +3962,17 @@ class WebRTCPeerSession:
         """
         url = msg.get("url")
         key = url if isinstance(url, str) else ""
+
+        # Checked before the cache, not after: the operator turning previews
+        # off has to stop serving the ones already fetched too, or the setting
+        # takes effect only for links nobody has posted yet. Refused as an
+        # ordinary miss — the client shows the bare link, which is exactly what
+        # "no preview" looks like for a page that has none.
+        if not self._group_ctx().get("chat_link_preview", True):
+            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
+                        "url": key, "ok": False})
+            return
+
         cached = _link_preview_cache_get(key)
         if cached is not None:
             self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION})
@@ -3678,51 +4031,101 @@ class WebRTCPeerSession:
                         "filename": filename})
             return
 
-        # The operator can close uploading to everyone but themselves. Enforced
-        # here rather than by hiding a button: the button is a courtesy to the
-        # people who are not trying, and this is the part that holds against
-        # someone who is. `is_node_admin` is computed from the identity this
-        # node pinned, never from a hub claim.
-        if not ctx.get("member_upload", True) and not self._is_node_admin():
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
             self._send({"type": "error",
-                        "detail": "Uploading is turned off for this group",
-                        "code": "member_upload_off",
+                        "detail": "No directories configured for this group",
                         "filename": filename})
-            self._audit("upload_refused", filename[:64])
             return
 
-        roots: RootSet | None = ctx.get("roots")
-        upload_root = roots.upload_root if roots else None
+        # The client names the root it is uploading into — it is browsing one,
+        # and with several writable roots any other choice is a guess. It names
+        # a root, never a path: the destination inside it is decided below and
+        # is not negotiable, which is what keeps C5a closed.
+        #
+        # An unknown name is refused rather than falling back to a writable
+        # root, because "the file went somewhere else" is discovered weeks
+        # later — the same reason the old single upload root was never guessed.
+        # A client that names nothing is an MNP 1.0 one, and there was exactly
+        # one destination in its world: the first writable root.
+        # `dir` is the folder being browsed, as a virtual path
+        # (`Media/Films/1999`); `root` is the older, coarser form and is what
+        # its first segment means on its own.
+        target_rel = str(msg.get("dir") or "").strip().strip("/")
+        target_root_name = (target_rel.split("/")[0] if target_rel
+                            else str(msg.get("root") or "").strip())
+        upload_root = None
+        if target_root_name:
+            upload_root = roots.by_name(target_root_name)
+            if upload_root is None:
+                self._send({"type": "error",
+                            "detail": f"No directory named "
+                                      f"{target_root_name!r} in this group",
+                            "code": "no_such_root",
+                            "filename": filename})
+                return
+        else:
+            writable = roots.writable_roots
+            upload_root = writable[0] if writable else None
+
         if upload_root is None:
-            # Refused, never guessed. With several roots, picking one would send
-            # a member's file to a disk the operator did not intend, and that is
-            # discovered weeks later.
             self._send({"type": "error",
-                        "detail": "No upload folder is configured for this group",
+                        "detail": "No writable directory in this group",
+                        "code": "no_writable_root",
+                        "filename": filename})
+            return
+        if not upload_root.writable:
+            self._send({"type": "error",
+                        "detail": f"Directory '{upload_root.name}' is read-only",
+                        "code": "root_read_only",
                         "filename": filename})
+            self._audit("upload_refused", filename[:64])
             return
         if not upload_root.available:
-            # The designated root's volume is absent. Falling back to another
-            # root would scatter uploads across disks depending on what happened
-            # to be plugged in.
             self._send({"type": "error",
-                        "detail": f"The upload folder ({upload_root.name}) is "
+                        "detail": f"Directory '{upload_root.name}' is "
                                   f"currently unavailable",
+                        "code": "root_unavailable",
                         "filename": filename})
             return
 
-        if upload_root.direct:
-            rel_dir = upload_root.name
-            target_dir = upload_root.path
+        # The folder the sender is looking at, and no subdirectory of the node's
+        # invention.
+        #
+        # Uploads used to be confined to `<root>/uploads/`, created on demand.
+        # That was the last of v5's quarantine (the per-user layer went on
+        # 2026-08-14, for the same reason): a shared directory nobody can
+        # organise is not a shared directory, and a folder appearing beside the
+        # operator's library because somebody sent a file is the node deciding
+        # how their disk is arranged.
+        #
+        # What made the quarantine worth having is not the subdirectory — it is
+        # the filename allowlist, the size cap, the chunk ordering, and the
+        # no-overwrite rule below. All four are unchanged.
+        #
+        # `resolve()` and not a join: it refuses `..`, absolute segments and
+        # anything whose resolved form escapes its root, symlinks included. The
+        # client names *where among the group's own folders*, never a path on
+        # the operator's filesystem.
+        if target_rel:
+            target_dir = roots.resolve(target_rel)
+            if target_dir is None or not target_dir.is_dir():
+                self._send({"type": "error",
+                            "detail": "Not a directory in this group",
+                            "code": "no_such_directory",
+                            "filename": filename})
+                return
+            rel_dir = target_rel
         else:
-            rel_dir = f"{upload_root.name}/{UPLOAD_DIR_NAME}"
-            target_dir = upload_root.path / UPLOAD_DIR_NAME
-            try:
-                target_dir.mkdir(parents=True, exist_ok=True)
-            except OSError as e:
-                log.warning("Cannot create upload folder in root %r: %s",
-                            upload_root.name, e)
-                self._send({"type": "error", "detail": "Upload folder unavailable",
+            # An MNP 1.0 client names nothing; the root itself is where its one
+            # destination now is.
+            target_dir = upload_root.path
+            rel_dir = upload_root.name
+            if not target_dir.is_dir():
+                self._send({"type": "error",
+                            "detail": f"Directory '{upload_root.name}' is "
+                                      f"currently unavailable",
+                            "code": "root_unavailable",
                             "filename": filename})
                 return
 
@@ -3981,8 +4384,10 @@ class WebRTCPeerSession:
             self._spawn(
                 self._admin_exec_member_unpin(pending, transcript, sig_bytes))
         elif pending["op"] == OP_MEMBER_UPLOAD:
-            self._spawn(
-                self._admin_exec_member_upload(pending, transcript, sig_bytes))
+            log.warning("Deprecated OP_MEMBER_UPLOAD signed op — use root "
+                        "writable/read-only instead")
+            self._send({"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
+                        "allowed": True, "deprecated": True})
         elif pending["op"] == OP_APPS_ENABLED:
             self._spawn(
                 self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
@@ -4019,6 +4424,24 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_ROOT_REMOVE:
             self._spawn(
                 self._admin_exec_root_remove(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_APP_DIRECTORIES:
+            self._spawn(
+                self._admin_exec_app_directories(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_CHAT_DIRECTORY:
+            self._spawn(
+                self._admin_exec_chat_directory(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_CHAT_LINK_PREVIEW:
+            self._spawn(
+                self._admin_exec_chat_link_preview(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_ROOT_UPDATE:
+            self._spawn(
+                self._admin_exec_root_update(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_ROOT_EJECT:
+            self._spawn(
+                self._admin_exec_root_eject(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_ROOT_PLUG:
+            self._spawn(
+                self._admin_exec_root_plug(pending, transcript, sig_bytes))
         elif pending["op"] == OP_GROUP_ATTACH:
             self._spawn(
                 self._admin_exec_group_attach(pending, transcript, sig_bytes))
diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py
index c683204..6986b01 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/wire.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py
@@ -89,13 +89,21 @@ def index_sync_message(index, roots: RootSet | None) -> dict:
     }
 
 
-def index_delta_message(index, delta) -> dict:
+def index_delta_message(index, delta, roots=None) -> dict:
     """
     One `index_delta` — what changed since the last thing this node broadcast.
 
     Built here rather than inline in the daemon, which is where it lived and which
     made it the third place an index message was constructed: precisely the drift
     that produced two `index_sync` encodings and two `file_chunk` encodings before it.
+
+    `roots` rides along (MNP 1.1, additive — a 1.0 client ignores it). It used
+    to travel on `index_sync` alone, which is a *full* index and therefore only
+    ever sent on request. So a root added, removed, ejected or plugged left
+    every connected client's directory table stale until somebody reloaded the
+    page: the delta that told them something had changed was the one message
+    that could not say what. It is a handful of dicts, bounded by the number of
+    directories a group has, and it is sealed with the rest.
     """
     payload = {
         "base_version": delta.base_version,
@@ -104,6 +112,8 @@ def index_delta_message(index, delta) -> dict:
         "deletions": list(delta.deletions),
         "updates": [index_entry_wire(e) for e in delta.updates],
     }
+    if roots is not None:
+        payload["roots"] = roots.describe()
     return {
         "type": MNP.INDEX_DELTA,
         "v": MNP_VERSION,
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index 6fdc78f..fc6c04a 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -175,7 +175,7 @@ def create_ui_app(state: dict) -> FastAPI:
             state,
             (payload.get("name") or "").strip(),
             (payload.get("shared_dir") or "").strip(),
-            upload_dir=(payload.get("upload_dir") or "").strip(),
+            writable=bool(payload.get("writable", True)),
         ))
         reload_fn = state.get("reload_fn")
         if reload_fn:
@@ -349,13 +349,35 @@ def create_ui_app(state: dict) -> FastAPI:
             (payload.get("path") or "").strip(),
             name=(payload.get("name") or "").strip(),
             kind=(payload.get("kind") or "generic").strip(),
-            upload=bool(payload.get("upload", False)),
+            writable=bool(payload.get("writable",
+                                      payload.get("upload", False))),
+            removable=bool(payload.get("removable", False)),
         ))
         reload_fn = state.get("reload_fn")
         if reload_fn:
             asyncio.ensure_future(reload_fn())
         return result
 
+    @app.patch("/api/groups/{group_id}/roots/{root_name}")
+    async def update_root(group_id: str, root_name: str, payload: dict):
+        result = await _op(lambda: ops.update_root(
+            state, group_id, root_name,
+            writable=payload.get("writable"),
+            removable=payload.get("removable"),
+        ))
+        reload_fn = state.get("reload_fn")
+        if reload_fn:
+            asyncio.ensure_future(reload_fn())
+        return result
+
+    @app.put("/api/groups/{group_id}/roots/{root_name}/eject")
+    async def eject_root(group_id: str, root_name: str):
+        return await _op(lambda: ops.eject_root(state, group_id, root_name))
+
+    @app.put("/api/groups/{group_id}/roots/{root_name}/plug")
+    async def plug_root(group_id: str, root_name: str):
+        return await _op(lambda: ops.plug_root(state, group_id, root_name))
+
     @app.delete("/api/groups/{group_id}/roots/{root_name}")
     async def remove_root(group_id: str, root_name: str):
         result = await _op(lambda: ops.remove_root(state, group_id, root_name))
@@ -397,14 +419,6 @@ def create_ui_app(state: dict) -> FastAPI:
             "current_dir": progress.current_dir,
         }
 
-    # ── Upload toggle (operator only, localhost) ─────────────────────────
-
-    @app.put("/api/groups/{group_id}/member-upload")
-    async def set_member_upload(group_id: str, payload: dict):
-        return await _op(lambda: ops.set_member_upload(
-            state, group_id, bool(payload.get("allowed", False)),
-        ))
-
     # ── Enabled apps (operator only, localhost) ────────────────────────────
     #
     # Same loopback shape as member-upload: the Create Group wizard sets this
@@ -419,6 +433,33 @@ def create_ui_app(state: dict) -> FastAPI:
             raise HTTPException(400, "apps must be a non-empty list")
         return await _op(lambda: ops.set_enabled_apps(state, group_id, apps))
 
+    # ── App directories (operator only, localhost) ────────────────────────
+    #
+    # The loopback twin of the `app_directories` MNP op. One endpoint for every
+    # application, keyed by the app's own name, so adding one needs no route
+    # here — the same reason the op is generic. `ALLOWED_APPS` is checked on
+    # the MNP path; here the caller is already on localhost holding the run
+    # token, and `ops` refuses a directory outside the group's roots either
+    # way, so an unknown key writes one unread settings row and nothing else.
+
+    @app.put("/api/groups/{group_id}/app-directories/{app_key}")
+    async def set_app_directories(group_id: str, app_key: str, payload: dict):
+        dirs = payload.get("directories")
+        if not isinstance(dirs, list):
+            raise HTTPException(400, "directories must be a list")
+        return await _op(lambda: ops.set_app_directories(
+            state, group_id, app_key, [str(d) for d in dirs]))
+
+    @app.put("/api/groups/{group_id}/chat-directory")
+    async def set_chat_directory(group_id: str, payload: dict):
+        return await _op(lambda: ops.set_chat_directory(
+            state, group_id, str(payload.get("path") or "")))
+
+    @app.put("/api/groups/{group_id}/chat-link-preview")
+    async def set_chat_link_preview(group_id: str, payload: dict):
+        return await _op(lambda: ops.set_chat_link_preview(
+            state, group_id, bool(payload.get("enabled", True))))
+
     # ── Scan settings (operator only, localhost) ──────────────────────────
 
     @app.put("/api/groups/{group_id}/scan-settings")
diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py
index 20724aa..3dc9cd9 100644
--- a/packages/meshbay-node/tests/conftest.py
+++ b/packages/meshbay-node/tests/conftest.py
@@ -24,14 +24,18 @@ win32_todo = pytest.mark.skipif(
 )
 
 
-def one_root(path: Path, *, name: str = "", kind: str = "generic") -> RootSet:
+def one_root(path: Path, *, name: str = "", kind: str = "generic",
+             writable: bool = True) -> RootSet:
     """
-    A RootSet with a single root over `path`, receiving uploads.
+    A RootSet with a single writable root over `path`.
 
     The equivalent of the old `shared_dir`. Note what it implies for assertions:
     a file directly in `path` now has `entry.path == <basename of path>`, not
     `""` — every index path carries its root name, in a group with one root as
     much as in a group with five.
+
+    Writable by default because most callers are testing something else and
+    want a root an upload can reach. `writable=False` is the read-only group.
     """
     return RootSet.build([{"path": str(path), "name": name, "kind": kind,
-                           "upload": True}])
+                           "writable": writable}])
diff --git a/packages/meshbay-node/tests/test_app_directories.py b/packages/meshbay-node/tests/test_app_directories.py
new file mode 100644
index 0000000..3ede1b6
--- /dev/null
+++ b/packages/meshbay-node/tests/test_app_directories.py
@@ -0,0 +1,292 @@
+"""
+One shape for every application's directories.
+
+`video_root` (a string), `audio_root` (a string) and `photo_roots` (a list)
+said the same thing three ways, and each needed its own op, its own MNP message
+and its own settings widget. They are one function keyed by the app's own name
+now, which is what lets an application be added without touching this layer at
+all — the whole claim of the plugin architecture.
+
+Two properties are new rather than moved, and both matter more than the tidying:
+
+* **the paths are validated.** The setters this replaces accepted anything. A
+  typo, or a path left behind when a root was removed, was stored happily and
+  then matched no entry — an app showing an empty tab, with nothing to
+  distinguish "misconfigured" from "no files yet". The moment of setting is the
+  only one where the operator is present to be told;
+* **the legacy scalar is derived, never stored.** `video_root` still rides on
+  the handshake ack for MNP 1.0 clients. Kept as a second stored value it would
+  drift from the list within one run — the shape of bug that reads as "it works
+  after a restart".
+"""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+async def _state(tmp_path: Path, *, writable: bool = True) -> tuple[dict, Roster]:
+    media = tmp_path / "Media"
+    (media / "Films").mkdir(parents=True)
+    (media / "Albums").mkdir()
+    published = tmp_path / "Published"
+    published.mkdir()
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roots = RootSet.build([
+        {"path": str(media), "writable": writable},
+        {"path": str(published)},
+    ])
+    state = {
+        "roster": roster,
+        "node_user_id": "operator",
+        "groups_ctx": {GROUP: {"index": index, "roots": roots}},
+        "config": SimpleNamespace(groups=[SimpleNamespace(id=GROUP, roots=[])]),
+    }
+    return state, roster
+
+
+# ── One function, any app ────────────────────────────────────────────────────
+
+async def test_an_app_nobody_wrote_code_for_stores_its_directories(tmp_path):
+    """
+    The point of the generic pair. Nothing in ops.py, roster.py or the daemon
+    names this app, and it round-trips anyway — which is the difference between
+    a plugin architecture and a list of special cases.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        await ops.set_app_directories(state, GROUP, "helloworld", ["Media/Films"])
+        assert await roster.app_directories(GROUP, "helloworld") == ["Media/Films"]
+    finally:
+        await roster.close()
+
+
+async def test_directories_are_deduplicated_and_ordered(tmp_path):
+    """
+    The stored form is what the operator signs a subject built from, on both
+    sides. Two clients sending the same set in different orders must produce
+    the same bytes, or one of them refuses to sign its own request.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        out = await ops.set_app_directories(
+            state, GROUP, "video", ["Media/Films", "Media", "Media/Films"])
+        assert out["directories"] == ["Media", "Media/Films"]
+    finally:
+        await roster.close()
+
+
+async def test_a_single_directory_app_stores_a_one_element_list(tmp_path):
+    state, roster = await _state(tmp_path)
+    try:
+        out = await ops.set_app_directory(state, GROUP, "chat", "Media/Films",
+                                          require_writable=True)
+        assert out["path"] == "Media/Films"
+        assert await roster.app_directories(GROUP, "chat") == ["Media/Films"]
+
+        cleared = await ops.set_app_directory(state, GROUP, "chat", "")
+        assert cleared["path"] == ""
+        assert await roster.app_directories(GROUP, "chat") == []
+    finally:
+        await roster.close()
+
+
+# ── Validation ───────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("bad", [
+    "Nowhere", "Nowhere/Deeper", "/etc", "Media/../../etc", "..",
+])
+async def test_a_directory_outside_every_root_is_refused(tmp_path, bad):
+    state, roster = await _state(tmp_path)
+    try:
+        with pytest.raises(ops.OpError):
+            await ops.set_app_directories(state, GROUP, "video", [bad])
+        assert await roster.app_directories(GROUP, "video") == []
+    finally:
+        await roster.close()
+
+
+async def test_a_read_only_root_is_refused_where_writability_is_required(tmp_path):
+    """
+    Chat's directory is a destination, not a view. Storing one on a read-only
+    root would produce a paperclip that fails at the moment somebody uses it,
+    which is the failure the RO/RW model exists to move earlier.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        with pytest.raises(ops.OpError, match="read-only"):
+            await ops.set_app_directory(state, GROUP, "chat", "Published",
+                                        require_writable=True)
+        # The same path is fine for an app that only reads it.
+        await ops.set_app_directories(state, GROUP, "video", ["Published"])
+        assert await roster.app_directories(GROUP, "video") == ["Published"]
+    finally:
+        await roster.close()
+
+
+async def test_a_directory_on_an_unplugged_drive_can_still_be_configured(tmp_path):
+    """
+    Deliberately *not* `RootSet.resolve()`, which also refuses a root that is
+    currently unavailable. An operator must be able to point an app at a
+    library on a drive they have ejected — what is checked is the shape, which
+    does not change with what happens to be mounted.
+    """
+    state, roster = await _state(tmp_path)
+    roots = state["groups_ctx"][GROUP]["roots"]
+    roots.roots[0].available = False
+    try:
+        out = await ops.set_app_directories(state, GROUP, "video", ["Media/Films"])
+        assert out["directories"] == ["Media/Films"]
+    finally:
+        await roster.close()
+
+
+# ── The derived scalar ───────────────────────────────────────────────────────
+
+async def test_the_legacy_scalar_follows_the_list_in_the_live_context(tmp_path):
+    """
+    `video_root` rides on the handshake ack for MNP 1.0 clients and is read
+    from the group context. Left behind by a save, it would disagree with the
+    list until the next restart.
+    """
+    state, roster = await _state(tmp_path)
+    ctx = state["groups_ctx"][GROUP]
+    try:
+        await ops.set_app_directories(state, GROUP, "video",
+                                      ["Media/Films", "Media/Albums"])
+        assert ctx["video_directories"] == ["Media/Albums", "Media/Films"]
+        assert ctx["video_root"] == "Media/Albums", (
+            "the scalar must be the first of the list, not a stale value")
+
+        await ops.set_app_directories(state, GROUP, "video", [])
+        assert ctx["video_root"] == ""
+    finally:
+        await roster.close()
+
+
+async def test_the_photo_alias_stays_a_list_and_chat_stays_a_string(tmp_path):
+    """The alias table has to carry the shape, not just the name."""
+    state, roster = await _state(tmp_path)
+    ctx = state["groups_ctx"][GROUP]
+    try:
+        await ops.set_app_directories(state, GROUP, "photo",
+                                      ["Media/Films", "Media/Albums"])
+        assert ctx["photo_roots"] == ["Media/Albums", "Media/Films"]
+        await ops.set_app_directory(state, GROUP, "chat", "Media",
+                                    require_writable=True)
+        assert ctx["chat_directory"] == "Media"
+    finally:
+        await roster.close()
+
+
+# ── Reading what an older node stored ────────────────────────────────────────
+
+async def test_an_existing_video_root_is_read_without_a_migration(tmp_path):
+    """
+    A node upgraded into this reads its old key until the first save through
+    the new path. Requiring a migration script to run before the Videos tab
+    works again would be a step nobody performs on the machine where it
+    matters.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "Media/Films")
+        assert await roster.app_directories(GROUP, "video") == ["Media/Films"]
+
+        await roster.set_setting(GROUP, "audio_root", "Media/Albums")
+        assert await roster.app_directories(GROUP, "music") == ["Media/Albums"]
+
+        await roster.set_setting(GROUP, "photo_roots", '["A", "B"]')
+        assert await roster.app_directories(GROUP, "photo") == ["A", "B"]
+    finally:
+        await roster.close()
+
+
+async def test_an_empty_legacy_value_means_nothing_configured(tmp_path):
+    """`video_root = ""` was how "unset" was spelled; it must not become `[""]`."""
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "")
+        assert await roster.app_directories(GROUP, "video") == []
+    finally:
+        await roster.close()
+
+
+async def test_the_new_key_wins_over_the_legacy_one(tmp_path):
+    """
+    Both present is a group saved once through the new path. Reading the old
+    key there would undo that save on every load.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "Old/Place")
+        await roster.set_app_directories(GROUP, "video", ["New/Place"])
+        assert await roster.app_directories(GROUP, "video") == ["New/Place"]
+    finally:
+        await roster.close()
+
+
+async def test_an_app_with_no_legacy_name_simply_has_none(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.app_directories(GROUP, "helloworld") == []
+    finally:
+        await roster.close()
+
+
+# ── Chat's own settings ──────────────────────────────────────────────────────
+
+async def test_link_previews_default_on_and_survive_a_restart(tmp_path):
+    """
+    Absent means on, because that is what the node did before the switch
+    existed — an upgrade must not silently change what a group's chat does.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.chat_link_preview(GROUP) is True
+        await roster.set_chat_link_preview(GROUP, False, set_by="op")
+        assert await roster.chat_link_preview(GROUP) is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.chat_link_preview(GROUP) is False
+        assert await reopened.chat_link_preview("other") is True, (
+            "one group's setting must not answer for another")
+    finally:
+        await reopened.close()
+
+
+async def test_setting_link_previews_updates_the_live_context(tmp_path):
+    """
+    The unfurl handler reads the context, not the database — it runs per
+    message and a round trip there would be absurd. So the two are kept in
+    step by the op that changes it.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        await ops.set_chat_link_preview(state, GROUP, False)
+        assert state["groups_ctx"][GROUP]["chat_link_preview"] is False
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_apps_enabled_policy.py b/packages/meshbay-node/tests/test_apps_enabled_policy.py
index 671005a..ac44ab3 100644
--- a/packages/meshbay-node/tests/test_apps_enabled_policy.py
+++ b/packages/meshbay-node/tests/test_apps_enabled_policy.py
@@ -1,7 +1,7 @@
 """
 The operator decides which group "applications" (Chat, Files, ...) are shown.
 
-Same shape as `test_member_upload_policy.py`, because it is the same kind of
+Same shape as `test_root_writable_policy.py`, because it is the same kind of
 setting: changed by a signed operator instruction, stored on the node rather
 than the hub, and safe for an existing group to have never heard of. The two
 things specific to this one: the whole set is signed in one message rather
@@ -88,7 +88,7 @@ async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
 
 async def test_changing_it_needs_a_signature(tmp_path):
     """The request only ever produces a challenge. Nothing is applied until a
-    signature over the transcript verifies — the same path as member_upload."""
+    signature over the transcript verifies — the same path as the root ops."""
     session = _session(tmp_path, "op", operator="op")
     session._has_admin_authority = lambda: True
     issued = []
diff --git a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
index deab1bd..5a8f9b1 100644
--- a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
+++ b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
@@ -102,7 +102,7 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
 
     daemon = await _make_daemon(tmp_path, shared, group_id)
     try:
-        await daemon._roster.set_audio_root(group_id, "shared/Music", set_by="op")
+        await daemon._roster.set_app_directories(group_id, "music", ["shared/Music"], set_by="op")
         indexer = DirectoryIndexer(
             roots=one_root(shared), group_id=group_id,
             sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
@@ -142,8 +142,10 @@ async def test_setting_the_audio_root_sweeps_what_it_already_contains(tmp_path):
         # this file all along.
         state = {
             "roster": daemon._roster,
-            "groups_ctx": {group_id: {}},
-            "enrich_audio_root_fn": daemon._enrich_audio_root_now,
+            # Real roots, because ops now refuses a directory that is not
+            # inside one — the per-app setters this replaced validated nothing.
+            "groups_ctx": {group_id: {"roots": one_root(shared)}},
+            "enrich_app_dirs_fns": {"music": daemon._enrich_audio_root_now},
         }
         await ops.set_audio_root(state, group_id, "shared/Music")
         await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
@@ -197,8 +199,10 @@ async def test_the_same_file_shared_into_two_groups_enriches_in_both(tmp_path):
             "the fixture itself must produce identical content hashes — "
             "otherwise this test isn't exercising the collision at all")
 
-        await daemon._roster.set_audio_root(group_a, "shared_a/Music", set_by="op")
-        await daemon._roster.set_audio_root(group_b, "shared_b/Music", set_by="op")
+        await daemon._roster.set_app_directories(
+            group_a, "music", ["shared_a/Music"], set_by="op")
+        await daemon._roster.set_app_directories(
+            group_b, "music", ["shared_b/Music"], set_by="op")
 
         await daemon._enrich_new_audio_entries(indexer_a, list(indexer_a.index.entries))
         await daemon._enrich_new_audio_entries(indexer_b, list(indexer_b.index.entries))
diff --git a/packages/meshbay-node/tests/test_audio_root_policy.py b/packages/meshbay-node/tests/test_audio_root_policy.py
index 576c08a..e2e9254 100644
--- a/packages/meshbay-node/tests/test_audio_root_policy.py
+++ b/packages/meshbay-node/tests/test_audio_root_policy.py
@@ -125,17 +125,20 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        assert await roster.audio_root("g1") == "", "absent must mean unset"
-        await roster.set_audio_root("g1", "shared/Music", set_by="op")
-        assert await roster.audio_root("g1") == "shared/Music"
+        assert await roster.app_directories("g1", "music") == [], (
+            "absent must mean unset")
+        await roster.set_app_directories("g1", "music", ["shared/Music"],
+                                         set_by="op")
+        assert await roster.app_directories("g1", "music") == ["shared/Music"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.audio_root("g1") == "shared/Music"
-        assert await reopened.audio_root("g2") == "", "one group's setting must not answer for another"
+        assert await reopened.app_directories("g1", "music") == ["shared/Music"]
+        assert await reopened.app_directories("g2", "music") == [], (
+            "one group's setting must not answer for another")
     finally:
         await reopened.close()
 
@@ -239,7 +242,7 @@ async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_
 
         assert session._ctx["groups"][GROUP]["audio_root"] == "shared/Music", (
             "the live in-memory context must reflect the new root immediately")
-        assert await roster.audio_root(GROUP) == "shared/Music", (
+        assert await roster.app_directories(GROUP, "music") == ["shared/Music"], (
             "the same Roster instance must read back what it just wrote")
     finally:
         await roster.close()
@@ -249,7 +252,7 @@ async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_
     # actually answers "does it survive a reload".
     reopened = await open_roster(tmp_path)
     try:
-        assert await reopened.audio_root(GROUP) == "shared/Music", (
+        assert await reopened.app_directories(GROUP, "music") == ["shared/Music"], (
             "a freshly-opened Roster against the same db file must see the "
             "committed value — anything else means the write was never "
             "durable in the first place")
diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
index f58c020..cf91564 100644
--- a/packages/meshbay-node/tests/test_cli_dispatch.py
+++ b/packages/meshbay-node/tests/test_cli_dispatch.py
@@ -26,6 +26,15 @@ VERBS = [
     ["status"],
     ["group", "list"],
     ["group", "add"],                      # missing --dir: usage, then exit
+    ["group", "add", "g", "--dir", "/tmp/media", "--no-writable"],
+    ["root", "list"],
+    ["root", "add"],                       # missing path: usage, then exit
+    ["root", "add", "/tmp/media", "--writable", "--removable"],
+    ["root", "remove", "media", "--yes"],
+    ["root", "set", "media", "--no-writable"],
+    ["root", "set", "media"],              # nothing to change: usage, then exit
+    ["root", "eject", "media"],
+    ["root", "plug", "media"],
     ["gek", "init"],
     ["gek", "rotate", "--yes"],
     ["gek-init"],
@@ -33,6 +42,9 @@ VERBS = [
     ["member", "invite", "bob"],
     ["member", "revoke", "bob"],
     ["member", "unpin", "bob"],
+    # Removed, and it has to say so rather than offering a username for a verb
+    # that no longer takes one.
+    ["member", "upload"],
     ["operator", "pair"],
     ["file", "list"],
     ["file", "rm", "abc", "--yes"],
@@ -224,3 +236,57 @@ def test_a_bare_invocation_with_no_config_yet_exits_cleanly(monkeypatch, tmp_pat
     assert "meshbay-node init" in capsys.readouterr().out
     assert not missing_config.parent.exists(), (
         "a fresh, unprovisioned start must not create anything on disk")
+
+
+def test_a_removed_verb_says_what_replaced_it():
+    """
+    `member upload` used to set a group-wide switch that no longer exists. It
+    reached the usage line for the *other* member verbs — "usage: meshbay-node
+    member upload <username>" — which advertises a removed feature and sends
+    the operator looking for a username it would then reject.
+
+    Naming it costs three lines and is the difference between an operator
+    finding `root set --writable` and concluding the CLI is broken.
+    """
+    import inspect
+    source = inspect.getsource(daemon_mod.main)
+    start = source.index('if args.command == "member":')
+    block = source[start:source.index('if args.command == "group":', start)]
+
+    assert 'sub == "upload"' in block, (
+        "`member upload` falls through to the generic usage line")
+    guidance = block[block.index('sub == "upload"'):]
+    guidance = guidance[:guidance.index("sys.exit")]
+    assert "root set" in guidance and "--writable" in guidance, (
+        "the message does not name what replaced it")
+
+
+def test_there_is_no_way_to_create_a_group_in_the_old_shape():
+    """
+    `--upload-dir` is gone, and documenting it as deprecated was the wrong
+    answer — which is what it got at first.
+
+    It wrote `upload_dir` into a brand-new `[[groups]]` block, and
+    `GroupConfig.__post_init__` reads that by forcing *every other root
+    read-only* and appending that path as the one writable one. So
+    `group add --dir X --writable --upload-dir Y` silently made X read-only:
+    two mechanisms deciding which directories accept uploads, one of them
+    invisible, in a group created after the model that replaced it.
+
+    Reading it stays — an existing node.toml must keep working, and that is the
+    only legitimate use. Writing it does not.
+    """
+    import inspect
+    source = inspect.getsource(daemon_mod.main)
+    assert "--upload-dir" not in source, (
+        "the CLI can still create a group in the pre-RO/RW shape")
+
+    from meshbay_node import ops
+    params = inspect.signature(ops.attach_group).parameters
+    assert "upload_dir" not in params, (
+        "attach_group still writes the legacy key")
+
+    # The read path is deliberately untouched.
+    from meshbay_node.config import GroupConfig
+    assert "upload_dir" in inspect.getsource(GroupConfig), (
+        "an existing node.toml using upload_dir would stop working")
diff --git a/packages/meshbay-node/tests/test_index_delta_carries_roots.py b/packages/meshbay-node/tests/test_index_delta_carries_roots.py
new file mode 100644
index 0000000..227f2d0
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_delta_carries_roots.py
@@ -0,0 +1,131 @@
+"""
+The message that says something changed has to be able to say what.
+
+A group's directory table travelled on `index_sync` alone — a *full* index,
+which the node only ever sends on request. Every ongoing change went out as an
+`index_delta`, which carried files and nothing else. So a root added, removed,
+ejected or plugged by the operator reached every other client's screen only
+when somebody happened to reload the page.
+
+It was hidden by the acks: `root_add_ack` and friends broadcast the new table
+to whoever was connected, so the common cases looked fine. What that could not
+cover is a client connecting mid-change, one whose ack was lost, or — the one
+that surfaced it — the operator's own client, where the ack landed and was then
+overwritten by an index fetched before the node had rebuilt anything.
+
+Additive on the wire (MNP 1.1): a 1.0 client sees a field it does not read.
+"""
+
+from pathlib import Path
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_common.groupbox import PURPOSE_INDEX, unseal
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.transport.wire import index_delta_message, index_sync_message
+
+
+GROUP = "g" * 32
+
+
+def _roots(tmp_path: Path) -> RootSet:
+    for name in ("Films", "Albums"):
+        (tmp_path / name).mkdir()
+    return RootSet.build([
+        {"path": str(tmp_path / "Films"), "writable": True},
+        {"path": str(tmp_path / "Albums"), "removable": True},
+    ])
+
+
+def _index() -> GroupIndex:
+    return GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate(),
+                      gek=generate_gek())
+
+
+def _payload(msg: dict, index: GroupIndex) -> dict:
+    """What a member actually reads, through the seal rather than around it."""
+    return unseal(index.gek, PURPOSE_INDEX, msg["type"], GROUP, msg)
+
+
+class _Delta:
+    base_version = 1
+    version = 2
+    additions: list = []
+    deletions: list = []
+    updates: list = []
+
+
+def test_a_delta_carries_the_directory_table(tmp_path):
+    index = _index()
+    msg = index_delta_message(index, _Delta(), _roots(tmp_path))
+    payload = _payload(msg, index)
+
+    assert [r["name"] for r in payload["roots"]] == ["Films", "Albums"]
+    assert payload["roots"][0]["writable"] is True
+    assert payload["roots"][1]["removable"] is True
+
+
+def test_the_table_is_sealed_with_the_rest(tmp_path):
+    """
+    It is group content, not routing. Only `type`, `v` and `group_id` stay in
+    clear, because a receiver has to route and authenticate before it would
+    trust a decryption.
+    """
+    index = _index()
+    msg = index_delta_message(index, _Delta(), _roots(tmp_path))
+    assert set(msg) - {"type", "v", "group_id"}, "nothing was sealed"
+    assert "roots" not in msg, "the directory table is outside the envelope"
+
+
+def test_a_delta_still_works_without_a_table(tmp_path):
+    """
+    The argument is optional, so an older caller — or a path that has no root
+    set to hand — produces a message a client reads exactly as before.
+    """
+    index = _index()
+    payload = _payload(index_delta_message(index, _Delta()), index)
+    assert "roots" not in payload
+    assert payload["version"] == 2
+
+
+def test_the_table_says_the_same_thing_on_both_messages(tmp_path):
+    """
+    Two encodings of one idea is the drift `wire.py` exists to prevent — it
+    already happened twice, for `index_sync` and for `file_chunk`.
+    """
+    index = _index()
+    roots = _roots(tmp_path)
+    delta = _payload(index_delta_message(index, _Delta(), roots), index)
+    sync = _payload(index_sync_message(index, roots), index)
+    assert delta["roots"] == sync["roots"]
+
+
+def test_the_table_never_carries_a_path(tmp_path):
+    """
+    This message goes to every member. Where a directory lives on the
+    operator's disk is theirs — see test_root_paths_are_operator_only.py.
+    """
+    index = _index()
+    payload = _payload(index_delta_message(index, _Delta(), _roots(tmp_path)),
+                       index)
+    assert not any("path" in r for r in payload["roots"])
+
+
+def test_an_ejected_root_is_visible_in_the_delta(tmp_path):
+    """
+    The case this was written for. An eject changes no file — the entries
+    freeze — so the delta it produces is empty of additions, deletions and
+    updates. Without the table it says literally nothing, which is how a
+    library disappearing from under the group's feet went unannounced.
+    """
+    index = _index()
+    roots = _roots(tmp_path)
+    roots.roots[1].ejected = True
+    roots.roots[1].available = False
+
+    payload = _payload(index_delta_message(index, _Delta(), roots), index)
+    assert payload["additions"] == [] and payload["deletions"] == []
+    assert payload["roots"][1]["ejected"] is True
+    assert payload["roots"][1]["available"] is False
diff --git a/packages/meshbay-node/tests/test_member_upload_policy.py b/packages/meshbay-node/tests/test_member_upload_policy.py
deleted file mode 100644
index b1dc0cb..0000000
--- a/packages/meshbay-node/tests/test_member_upload_policy.py
+++ /dev/null
@@ -1,176 +0,0 @@
-"""
-The operator can close uploading to everyone but themselves.
-
-The point of these tests is the difference between a hidden button and a closed
-door. The interface stops offering the control, which is a courtesy to the
-people who are not trying; **the node refuses the upload**, which is the part
-that holds against someone who is. A member who kept an old tab open, or who
-speaks MNP directly, gets the same answer as everyone else.
-
-Two further things are worth holding:
-
-* the setting is changed by a **signed** operator instruction. A node that took
-  it from an unsigned message would let any member turn it back on, and the
-  control would be a suggestion;
-* it is stored on the **node**, not the hub. A hub that could decide who may
-  write to the operator's disk is a hub with authority over the node, which is
-  the thing this whole design is arranged to avoid.
-"""
-
-import base64
-from pathlib import Path
-
-import pytest
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-
-from meshbay_common.adminop import OP_MEMBER_UPLOAD
-from meshbay_node.indexer.group_index import GroupIndex
-from meshbay_node.roster import Roster
-from meshbay_node.transport.webrtc_server import WebRTCPeerSession
-
-from conftest import one_root
-
-pytestmark = pytest.mark.asyncio
-
-
-def _session(tmp_path: Path, user_id: str, *, member_upload: bool,
-             operator: str | None = None) -> WebRTCPeerSession:
-    shared_root = tmp_path / "shared"
-    shared_root.mkdir(exist_ok=True)
-    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
-    ctx = {
-        "roots": one_root(shared_root),
-        "index": index,
-        "sk_node": index.sk_node,
-        "member_upload": member_upload,
-        "node_user_id": operator,
-    }
-    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
-    session._ctx = ctx
-    session._group_id = None
-    session._user_id = user_id
-    session._pk_user = ""
-    session._uploads = {}
-    session.sent = []
-    session._send = session.sent.append
-    session._audit = lambda *a, **k: None
-    return session
-
-
-def _upload(session, filename="clip.mp4", body=b"bytes"):
-    session._do_file_upload({
-        "filename": filename, "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(body).decode(),
-    })
-
-
-def _uploads_dir(session) -> Path:
-    return session._ctx["roots"].upload_root.path / "uploads"
-
-
-# ── The door, not the button ────────────────────────────────────────────────
-
-async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path):
-    session = _session(tmp_path, "member-1", member_upload=False,
-                       operator="the-operator")
-    _upload(session)
-
-    assert not (_uploads_dir(session) / "clip.mp4").exists(), (
-        "the file was written even though uploading is off — the setting is "
-        "decorative and the hidden button was the whole control")
-    refusal = [m for m in session.sent if m.get("type") == "error"]
-    assert refusal and refusal[0].get("code") == "member_upload_off"
-
-
-async def test_the_operator_can_still_upload(tmp_path):
-    """Otherwise turning it off locks the operator out of their own node, and
-    the only way back is a config file and a restart."""
-    session = _session(tmp_path, "the-operator", member_upload=False,
-                       operator="the-operator")
-    _upload(session)
-
-    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
-
-
-async def test_members_upload_normally_when_it_is_on(tmp_path):
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    _upload(session)
-
-    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
-
-
-async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path):
-    """An existing node's context has no such key. The absence must read as
-    "allowed", or upgrading the node silently closes every group."""
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    del session._ctx["member_upload"]
-    _upload(session)
-
-    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
-
-
-# ── Who may change it ───────────────────────────────────────────────────────
-
-async def test_changing_it_needs_a_signature(tmp_path):
-    """
-    The request only ever produces a challenge. Nothing is applied until a
-    signature over the transcript verifies — the same path as removing a member.
-    """
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    session._has_admin_authority = lambda: True
-    issued = []
-    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
-
-    session._do_member_upload({"allowed": False})
-
-    assert issued == [(OP_MEMBER_UPLOAD, "off")]
-    assert session._ctx["member_upload"] is True, "applied before it was signed"
-
-
-async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
-    """The operator is shown the subject before signing. "member_upload" tells
-    them nothing; "off" tells them what they are about to do."""
-    session = _session(tmp_path, "op", member_upload=False, operator="op")
-    session._has_admin_authority = lambda: True
-    issued = []
-    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
-
-    session._do_member_upload({"allowed": True})
-
-    assert issued == [(OP_MEMBER_UPLOAD, "on")]
-
-
-async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    session._has_admin_authority = lambda: False
-
-    session._do_member_upload({"allowed": False})
-
-    assert [m for m in session.sent if m.get("type") == "error"]
-
-
-# ── Where it is stored ──────────────────────────────────────────────────────
-
-async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
-    roster = Roster(db_path=tmp_path / "roster.db")
-    await roster.open()
-    try:
-        assert await roster.member_upload_allowed("g1") is True, (
-            "absent must mean allowed, or an upgrade closes every group")
-        await roster.set_member_upload("g1", False, set_by="op")
-        assert await roster.member_upload_allowed("g1") is False
-    finally:
-        await roster.close()
-
-    reopened = Roster(db_path=tmp_path / "roster.db")
-    await reopened.open()
-    try:
-        assert await reopened.member_upload_allowed("g1") is False
-        assert await reopened.member_upload_allowed("g2") is True, (
-            "one group's setting must not answer for another")
-    finally:
-        await reopened.close()
diff --git a/packages/meshbay-node/tests/test_node_status.py b/packages/meshbay-node/tests/test_node_status.py
index b56eb6e..091b1db 100644
--- a/packages/meshbay-node/tests/test_node_status.py
+++ b/packages/meshbay-node/tests/test_node_status.py
@@ -255,7 +255,7 @@ async def test_add_root_creates_directory_and_returns_info(tmp_path):
 
     from meshbay_node.config import NodeConfig, GroupConfig, RootSpec
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(shared), name="shared", kind="generic", upload=True),
+        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
     ])
 
     conf = tmp_path / "node.toml"
@@ -295,7 +295,7 @@ async def test_remove_root_requires_at_least_one_remaining(tmp_path):
 
     from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(shared), name="shared", kind="generic", upload=True),
+        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
     ])
     node_cfg = NodeConfig.__new__(NodeConfig)
     node_cfg.groups = [cfg]
@@ -314,16 +314,22 @@ async def test_remove_root_requires_at_least_one_remaining(tmp_path):
         await ops.remove_root(state, GROUP, "shared")
 
 
-async def test_remove_root_refuses_upload_root(tmp_path):
-    d1 = tmp_path / "uploads"
+async def test_removing_a_writable_root_is_allowed(tmp_path):
+    """
+    It used to be refused: with one designated upload root, removing it left
+    the group with nowhere to put an upload and no way to say so. Several roots
+    can be writable now, and a group with none is a valid read-only group — so
+    the refusal would be protecting a state that is no longer special.
+    """
+    d1 = tmp_path / "incoming"
     d2 = tmp_path / "shared"
     d1.mkdir()
     d2.mkdir()
 
     from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(d1), name="uploads", kind="generic", upload=True),
-        RootSpec(path=str(d2), name="shared", kind="generic", upload=False),
+        RootSpec(path=str(d1), name="incoming", kind="generic", writable=True),
+        RootSpec(path=str(d2), name="shared", kind="generic", writable=False),
     ])
     node_cfg = NodeConfig.__new__(NodeConfig)
     node_cfg.groups = [cfg]
@@ -331,7 +337,7 @@ async def test_remove_root_refuses_upload_root(tmp_path):
     conf = tmp_path / "node.toml"
     conf.write_text(
         f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
-        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "uploads"\n  upload = true\n\n'
+        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "incoming"\n  writable = true\n\n'
         f'  [[groups.roots]]\n  path = "{d2}"\n  name = "shared"\n')
     roots = RootSet.build([asdict(r) for r in cfg.roots])
     index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
@@ -340,8 +346,97 @@ async def test_remove_root_refuses_upload_root(tmp_path):
         "config_path": str(conf),
         "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
     }
-    with pytest.raises(ops.OpError, match="upload root"):
-        await ops.remove_root(state, GROUP, "uploads")
+    result = await ops.remove_root(state, GROUP, "incoming")
+    assert result["status"] == "removed"
+    assert [r["name"] for r in result["roots"]] == ["shared"]
+    assert conf.read_text().count("[[groups.roots]]") == 1
+
+
+async def test_update_root_rewrites_the_flags_in_node_toml(tmp_path):
+    """
+    The flags live in the operator's config file, so they survive a restart —
+    and the file is hand-written and full of comments, so the change is a line
+    edit rather than a round trip through a TOML writer that would discard
+    every one of them.
+    """
+    d1 = tmp_path / "media"
+    d1.mkdir()
+
+    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
+    cfg = GroupConfig(id=GROUP, name="test", roots=[
+        RootSpec(path=str(d1), name="media", kind="generic", writable=False),
+    ])
+    node_cfg = NodeConfig.__new__(NodeConfig)
+    node_cfg.groups = [cfg]
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
+        f'  [[groups.roots]]\n'
+        f'  # the operator explained this one to themselves\n'
+        f'  path = "{d1}"\n  name = "media"\n')
+    roots = RootSet.build([asdict(r) for r in cfg.roots])
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    state = {
+        "config": node_cfg,
+        "config_path": str(conf),
+        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
+    }
+
+    result = await ops.update_root(state, GROUP, "media",
+                                   writable=True, removable=True)
+    assert result["status"] == "updated"
+    text = conf.read_text()
+    assert "writable = true" in text
+    assert "removable = true" in text
+    assert "the operator explained this one to themselves" in text, (
+        "the config file was rewritten instead of edited")
+
+    # And the live root set agrees immediately, without waiting for a reload:
+    # the loopback API reads it, and an operator who toggles a switch and sees
+    # it snap back assumes the change did not take.
+    assert roots.roots[0].writable is True
+    assert roots.roots[0].removable is True
+
+    # A second call that changes nothing must not append a duplicate line.
+    await ops.update_root(state, GROUP, "media", writable=True, removable=True)
+    assert conf.read_text().count("writable =") == 1
+
+
+async def test_update_root_replaces_a_legacy_upload_line(tmp_path):
+    """
+    A config written before the refactor says `upload = true`. Leaving it in
+    place next to a new `writable` line would give the file two answers, and
+    `RootSet.build` prefers `writable` — so the stale one would sit there
+    contradicting the running node for as long as anyone read it.
+    """
+    d1 = tmp_path / "media"
+    d1.mkdir()
+
+    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
+    cfg = GroupConfig(id=GROUP, name="test", roots=[
+        RootSpec(path=str(d1), name="media", kind="generic", writable=True),
+    ])
+    node_cfg = NodeConfig.__new__(NodeConfig)
+    node_cfg.groups = [cfg]
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
+        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "media"\n'
+        f'  upload = true\n')
+    roots = RootSet.build([asdict(r) for r in cfg.roots])
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    state = {
+        "config": node_cfg,
+        "config_path": str(conf),
+        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
+    }
+
+    await ops.update_root(state, GROUP, "media", writable=False)
+    text = conf.read_text()
+    assert "upload = true" not in text
+    assert "writable = false" in text
 
 
 async def test_remove_root_succeeds_with_two_roots(tmp_path):
@@ -352,8 +447,8 @@ async def test_remove_root_succeeds_with_two_roots(tmp_path):
 
     from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(d1), name="dir1", kind="generic", upload=True),
-        RootSpec(path=str(d2), name="dir2", kind="generic", upload=False),
+        RootSpec(path=str(d1), name="dir1", kind="generic", writable=True),
+        RootSpec(path=str(d2), name="dir2", kind="generic", writable=False),
     ])
     node_cfg = NodeConfig.__new__(NodeConfig)
     node_cfg.groups = [cfg]
diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py
index 92e32bf..c118b5a 100644
--- a/packages/meshbay-node/tests/test_ops.py
+++ b/packages/meshbay-node/tests/test_ops.py
@@ -11,12 +11,14 @@ call them.
 
 import asyncio
 import inspect
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
+from types import SimpleNamespace
 
 import pytest
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from meshbay_node import ops
 from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
 from meshbay_node.transport.quic_server import Denylist
 
 from conftest import one_root
@@ -64,8 +66,8 @@ def test_the_http_adapter_adds_no_logic():
     # Every endpoint that performs an operation routes through _op(...).
     for endpoint in ("operator_pair", "create_invite", "revoke_member",
                      "unpin_member", "init_gek", "attach_group", "delete_file",
-                     "add_root", "remove_root", "set_member_upload",
-                     "reload_config"):
+                     "add_root", "remove_root", "update_root",
+                     "eject_root", "plug_root", "reload_config"):
         start = source.index(f"async def {endpoint}(")
         body = source[start:start + 700]
         assert "_op(" in body.split("\n\n")[0] + body, (
@@ -181,25 +183,103 @@ async def test_an_unhosted_group_offers_what_it_does_host(tmp_path):
     assert exc.value.extra.get("available")
 
 
-# ── Upload policy (set_member_upload) ───────────────────────────────────────
+# ── Upload policy (per-root writable) ───────────────────────────────────────
 
-async def test_set_member_upload_toggles_and_persists(tmp_path):
+async def test_the_group_wide_upload_switch_is_gone(tmp_path):
+    """
+    `set_member_upload` was the whole of the old policy, and it is deliberately
+    not here any more — RO/RW on the root replaced it. A wrapper kept "for
+    compatibility" would be a second way to decide who writes to the operator's
+    disk, and two answers to that question is how C1 and C6 both happened.
+    """
+    assert not hasattr(ops, "set_member_upload")
+    from meshbay_node.roster import Roster
+    assert not hasattr(Roster, "set_member_upload")
+    assert not hasattr(Roster, "member_upload_allowed")
+
+
+async def test_eject_and_plug_persist_through_the_roster(tmp_path):
+    """
+    The state has to outlive the process: an operator ejects a drive, unplugs
+    it, and restarts the node — and the rescan that follows must not read the
+    empty mount point as an erased library.
+    """
     from meshbay_node.roster import Roster
     state = _state(tmp_path)
+    usb = tmp_path / "USB"
+    usb.mkdir()
+    state["groups_ctx"]["g" * 32]["roots"] = RootSet.build(
+        [{"path": str(usb), "removable": True, "writable": True}])
+    state["config"] = SimpleNamespace(
+        groups=[SimpleNamespace(id="g" * 32, roots=[])])
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     state["roster"] = roster
     state["node_user_id"] = "operator"
+    try:
+        out = await ops.eject_root(state, "g" * 32, "USB")
+        assert out["status"] == "ejected"
+        assert await roster.ejected_roots("g" * 32) == {"usb"}
+        assert out["roots"][0]["ejected"] is True
+        assert out["roots"][0]["available"] is False
 
-    out = await ops.set_member_upload(state, "g" * 32, True)
+        out = await ops.plug_root(state, "g" * 32, "USB")
+        assert out["status"] == "plugged"
+        assert await roster.ejected_roots("g" * 32) == set()
+    finally:
+        await roster.close()
 
-    assert out["allowed"] is True
-    assert state["groups_ctx"]["g" * 32]["member_upload"] is True
 
-    out2 = await ops.set_member_upload(state, "g" * 32, False)
+async def test_a_root_that_is_not_removable_cannot_be_ejected(tmp_path):
+    """
+    Eject means "I am about to unplug this". On a directory that is not on a
+    removable device it would hide a library with no way for the safety net to
+    notice anything happened, and nothing to plug back in.
+    """
+    from meshbay_node.roster import Roster
+    state = _state(tmp_path)
+    fixed = tmp_path / "Fixed"
+    fixed.mkdir()
+    state["groups_ctx"]["g" * 32]["roots"] = RootSet.build(
+        [{"path": str(fixed), "writable": True}])
+    state["config"] = SimpleNamespace(
+        groups=[SimpleNamespace(id="g" * 32, roots=[])])
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state["roster"] = roster
+    try:
+        with pytest.raises(ops.OpError, match="removable"):
+            await ops.eject_root(state, "g" * 32, "Fixed")
+    finally:
+        await roster.close()
 
-    assert out2["allowed"] is False
-    assert state["groups_ctx"]["g" * 32]["member_upload"] is False
+
+async def test_plugging_a_drive_that_is_not_there_is_refused(tmp_path):
+    """
+    Clearing the flag while the device is still absent would restart the
+    watchdog on a missing path and hand the next reconcile an empty directory —
+    the deletion storm the eject was there to prevent, produced by the recovery.
+    """
+    from meshbay_node.roster import Roster
+    state = _state(tmp_path)
+    usb = tmp_path / "USB"
+    usb.mkdir()
+    roots = RootSet.build([{"path": str(usb), "removable": True}])
+    roots.roots[0].ejected = True
+    roots.roots[0].available = False
+    state["groups_ctx"]["g" * 32]["roots"] = roots
+    state["config"] = SimpleNamespace(
+        groups=[SimpleNamespace(id="g" * 32, roots=[])])
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state["roster"] = roster
+    usb.rmdir()
+    try:
+        with pytest.raises(ops.OpError, match="device connected"):
+            await ops.plug_root(state, "g" * 32, "USB")
+        assert roots.roots[0].ejected is True
+    finally:
+        await roster.close()
 
 
 # ── Reload ──────────────────────────────────────────────────────────────────
@@ -279,13 +359,47 @@ def test_update_node_toml_forces_lf_and_keeps_standalone_comments(tmp_path):
 
 
 def test_a_backslash_path_written_into_node_toml_stays_parseable():
-    # attach_group / add_root / init embed a directory into a TOML basic string.
-    # A raw Windows path there (drive + backslash + "Users" + ...) is a parse
-    # error since backslash sequences are escapes; the code writes as_posix().
+    """
+    attach_group and add_root embed a directory into a TOML basic string. A raw
+    Windows path there is a parse error, because backslash sequences are escapes
+    (`\\U`, `\\a`, ...); the code writes `as_posix()` and pathlib reads `/` back
+    on Windows.
+
+    `PureWindowsPath`, not `Path`: on this suite's usual machine `Path` is a
+    `PosixPath`, where a backslash is an ordinary filename character and
+    `as_posix()` converts nothing — so the test modelled the wrong platform and
+    failed everywhere except the one it was written for. Naming the flavour
+    explicitly is what makes it the same assertion on all three.
+    """
     import tomllib
     bs = chr(92)
     win_dir = f"C:{bs}Users{bs}alice{bs}Media"
-    assert tomllib.loads(f'path = "{Path(win_dir).as_posix()}"\n')["path"] == \
-        "C:/Users/alice/Media"
+
+    assert tomllib.loads(
+        f'path = "{PureWindowsPath(win_dir).as_posix()}"\n'
+    )["path"] == "C:/Users/alice/Media"
+
     with pytest.raises(tomllib.TOMLDecodeError):
         tomllib.loads(f'path = "{win_dir}"\n')   # the bug this guards against
+
+
+def test_every_path_written_into_node_toml_goes_through_as_posix():
+    """
+    The half the round trip above cannot see.
+
+    Proving `as_posix()` produces a parseable string says nothing about whether
+    the code calls it, and this is a defect no Linux machine can reproduce: the
+    config is written, parsed and served correctly here, and fails on the
+    operator's Windows box. So the source is read for the shape instead —
+    weak evidence, and the only kind available for a platform the suite does
+    not run on.
+    """
+    import re
+    source = inspect.getsource(ops)
+    # Every f-string interpolation that lands on the right of a TOML `path =`.
+    writes = re.findall(r'path\s*=\s*\\?"\{([^}]+)\}', source)
+    assert writes, "no TOML path writer found — did the config writer move?"
+    for expr in writes:
+        assert "as_posix()" in expr, (
+            f'node.toml path written as `{expr}` — a Windows path needs '
+            f'as_posix(), or the file it lands in will not parse')
diff --git a/packages/meshbay-node/tests/test_rename_reenrichment.py b/packages/meshbay-node/tests/test_rename_reenrichment.py
index 7a77368..f6761d9 100644
--- a/packages/meshbay-node/tests/test_rename_reenrichment.py
+++ b/packages/meshbay-node/tests/test_rename_reenrichment.py
@@ -37,8 +37,8 @@ def _free_port() -> int:
 
 
 class _StubRoster:
-    async def video_root(self, group_id):
-        return "shared"
+    async def app_directories(self, group_id, app_key):
+        return ["shared"] if app_key == "video" else []
 
 
 class _SpyEnricher:
diff --git a/packages/meshbay-node/tests/test_replug_restores_enrichment.py b/packages/meshbay-node/tests/test_replug_restores_enrichment.py
new file mode 100644
index 0000000..04a06ae
--- /dev/null
+++ b/packages/meshbay-node/tests/test_replug_restores_enrichment.py
@@ -0,0 +1,318 @@
+"""
+A root that comes back keeps its Videos/Music/Photos metadata.
+
+Reported live: a removable root ejected from the Files app and plugged back
+in returned with its files and without its albums. Music showed "no music
+found", and it did not come back.
+
+`plug_root` has to re-walk the root — the drive may have changed while it
+was away — and `_scan_root` produces bare entries: `_hash_or_cached` fills
+id/name/path/size/type and nothing else. Every enrichment field went with
+the old object, and the Music tag fields are cached nowhere by design
+(`enrich_audio.py` re-reads them so a rename can re-derive the filename
+fallback).
+
+Two gates then stopped anything from filling them in again:
+
+* enrichment is scheduled for `delta.additions`, and ejecting broadcasts
+  nothing — the last snapshot still held those ids, so the rebuilt entries
+  diffed as *updates*;
+* `_enrich_new_*_entries` skips anything in `_enriched_attempted`, which is
+  only discarded for `delta.deletions` — and dropping and rescanning inside
+  one call broadcasts no deletion either.
+
+Only a restart cleared both, an empty snapshot making every entry an
+addition. That is why it looked like it might fix itself and never did.
+
+Re-enriching is now the *fallback*, not the fix. An entry's id is its
+content hash, so one that comes back under the same id, name and path is
+the same bytes in the same place and its enrichment still holds:
+`_rescan_root` carries those fields across. Re-deriving them instead meant
+tag reads, ffprobe runs and rate-limited lookups — measured at 14 seconds
+of empty Music tab on a real library with a cold cache, which to the
+operator is indistinguishable from the original bug.
+
+What these assert is therefore the field on the entry, not a call to an
+enricher. Counting calls is what made an earlier version of this file pass
+while the operator still watched their albums vanish.
+
+The same drop-and-rescan runs in `reconcile()` — "Root %r is back" — so a
+USB drive that falls off and returns on its own hits all of this without
+anybody touching the UI.
+"""
+
+import asyncio
+import os
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_node.config import (Config, GroupConfig, HubConfig, KeystoreConfig,
+                                 NodeConfig)
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.indexer.enrich_audio import AudioEnricher
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+# Above indexer.py's MIN_AUDIO_SIZE_BYTES, or nothing would be indexed.
+_AUDIO_BYTES = os.urandom(60 * 1024)
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+class _CountingEnricher:
+    """Stands in for an enricher: records who it was asked to enrich."""
+
+    def __init__(self):
+        self.spawned: list[str] = []
+
+    def spawn(self, entry, file_path, on_done, boundary=None):
+        self.spawned.append(entry.name)
+
+
+async def _daemon(tmp_path, shared, group_id):
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id=group_id, name="test-group", shared_dir=str(shared),
+            visibility="private", quic_port=_free_port(),
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
+    daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await daemon._media_cache.open()
+    daemon._roster = Roster(db_path=tmp_path / "roster.db")
+    await daemon._roster.open()
+    return daemon
+
+
+async def _settled(daemon, indexer):
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+
+async def _library(tmp_path, group_id, *, enricher=None):
+    """A one-track library under <root>/<artist>/<album>, enriched once."""
+    library = tmp_path / "music"
+    (library / "an artist" / "a record").mkdir(parents=True)
+    (library / "an artist" / "a record" / "01 first track.mp3").write_bytes(_AUDIO_BYTES)
+
+    daemon = await _daemon(tmp_path, library, group_id)
+    daemon._audio_enricher = enricher or AudioEnricher(daemon._media_cache)
+    await daemon._roster.set_app_directories(
+        group_id, "music", ["music"], set_by="op")
+
+    roots = RootSet.build([{"path": str(library), "removable": True}])
+    indexer = DirectoryIndexer(
+        roots=roots, group_id=group_id,
+        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+    await indexer.initial_scan()
+    await _settled(daemon, indexer)
+    await asyncio.sleep(0.4)  # the enricher runs off the broadcast
+    return daemon, indexer, roots, library
+
+
+def _only(indexer):
+    return next(iter(indexer.index.entries))
+
+
+# ── The operator's own eject and plug ────────────────────────────────────────
+
+async def test_the_albums_are_still_there_after_a_replug(tmp_path):
+    """
+    No tags are written: `enrich_audio._artist_album_from_ancestors` derives
+    artist and album from the folder names when a file has none, which is the
+    <library>/<artist>/<album>/<track> layout this was reported against.
+    """
+    group_id = "a" * 32
+    daemon, indexer, _, _ = await _library(tmp_path, group_id)
+    try:
+        before = _only(indexer)
+        assert before.album == "a record" and before.artist == "an artist", (
+            f"the first pass never filled the fields: {before}")
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        after = _only(indexer)
+        assert after.album == "a record" and after.artist == "an artist", (
+            "the entry came back from the rescan with no album — this is what "
+            "an empty Music tab after a replug looks like on the node")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+async def test_the_fields_survive_without_re_deriving_them(tmp_path):
+    """
+    Carried across, not recomputed. Re-deriving is correct and far too slow:
+    on a real library with a cold metadata cache it left Music empty for 14
+    seconds, and an operator who looks in that window sees the bug.
+    """
+    group_id = "b" * 32
+    enricher = _CountingEnricher()
+    daemon, indexer, _, _ = await _library(tmp_path, group_id, enricher=enricher)
+    try:
+        assert enricher.spawned == ["01 first track.mp3"]
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        assert enricher.spawned == ["01 first track.mp3"], (
+            "an unchanged file was enriched a second time — the whole point "
+            "of the content hash is that it did not need to be")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+async def test_who_uploaded_a_file_survives_it_too(tmp_path):
+    """
+    `uploader_id`/`uploader_pk` are the same shape of field — set once, on an
+    entry, readable from nowhere on disk — and they decide who may delete the
+    file. Losing them to a replug quietly takes a right away.
+    """
+    group_id = "c" * 32
+    daemon, indexer, _, _ = await _library(tmp_path, group_id)
+    try:
+        entry = _only(indexer)
+        entry.uploader_id = "alice"
+        entry.uploader_pk = "a-pinned-key"
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        after = _only(indexer)
+        assert after.uploader_id == "alice" and after.uploader_pk == "a-pinned-key"
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+# ── A drive that leaves and returns on its own ──────────────────────────────
+
+async def test_a_root_that_returns_on_its_own_is_treated_the_same(tmp_path):
+    """
+    `reconcile()` rescans a root that reappears without anyone asking — a USB
+    drive re-mounting. Same drop-and-rescan, so it lost the same fields, with
+    no click anywhere to blame it on.
+    """
+    group_id = "d" * 32
+    daemon, indexer, roots, _ = await _library(tmp_path, group_id)
+    try:
+        assert _only(indexer).album == "a record"
+
+        roots.roots[0].available = False
+        await indexer.reconcile()
+        await _settled(daemon, indexer)
+        await indexer.reconcile()
+        await _settled(daemon, indexer)
+        await asyncio.sleep(0.4)
+
+        assert _only(indexer).album == "a record", (
+            "a drive that fell off and came back left the library with no "
+            "metadata")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+# ── What genuinely does have to be re-derived ───────────────────────────────
+
+async def test_a_track_moved_while_the_drive_was_away_is_enriched_again(tmp_path):
+    """
+    The counter-case, and the reason the carry-over is keyed on name and path
+    as well as id. `artist`, `album`, `display_title` and `track_no` all fall
+    back to the folder and filename when a file carries no tags, so the same
+    bytes under a new name are not the same metadata. Those are the entries
+    the daemon still re-enriches, off `rescanned_ids`.
+    """
+    group_id = "e" * 32
+    enricher = _CountingEnricher()
+    daemon, indexer, _, library = await _library(
+        tmp_path, group_id, enricher=enricher)
+    try:
+        assert enricher.spawned == ["01 first track.mp3"]
+
+        moved = library / "another artist" / "another record"
+        moved.mkdir(parents=True)
+        (library / "an artist" / "a record" / "01 first track.mp3").rename(
+            moved / "01 first track.mp3")
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        assert enricher.spawned == ["01 first track.mp3"] * 2, (
+            "the file is under a different artist and album now; carrying the "
+            "old ones across would file it under a folder it left")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+async def test_videos_and_photos_are_covered_by_the_same_path(tmp_path):
+    """
+    Nothing here is specific to Music — Videos and Photos lost their durations,
+    titles and thumbnails the same way. Music is simply where it shows up
+    loudest: a track with no tags has no album to file it under, so the app
+    goes empty rather than merely plain.
+    """
+    group_id = "f" * 32
+    library = tmp_path / "media"
+    (library / "films").mkdir(parents=True)
+    (library / "films" / "clip.mkv").write_bytes(os.urandom(60 * 1024))
+    (library / "album").mkdir(parents=True)
+    (library / "album" / "shot.jpg").write_bytes(os.urandom(60 * 1024))
+
+    daemon = await _daemon(tmp_path, library, group_id)
+    video, photo = _CountingEnricher(), _CountingEnricher()
+    daemon._enricher, daemon._photo_enricher = video, photo
+    try:
+        await daemon._roster.set_app_directories(
+            group_id, "video", ["media/films"], set_by="op")
+        await daemon._roster.set_app_directories(
+            group_id, "photo", ["media/album"], set_by="op")
+
+        roots = RootSet.build([{"path": str(library), "removable": True}])
+        indexer = DirectoryIndexer(
+            roots=roots, group_id=group_id,
+            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+        await indexer.initial_scan()
+        await _settled(daemon, indexer)
+        assert video.spawned == ["clip.mkv"] and photo.spawned == ["shot.jpg"]
+
+        by_name = {e.name: e for e in indexer.index.entries}
+        by_name["clip.mkv"].duration = 1234
+        by_name["shot.jpg"].thumb_hash = "a-thumbnail"
+
+        indexer.eject_root("media")
+        await indexer.plug_root("media")
+        await _settled(daemon, indexer)
+
+        back = {e.name: e for e in indexer.index.entries}
+        assert back["clip.mkv"].duration == 1234, "the film lost its probe"
+        assert back["shot.jpg"].thumb_hash == "a-thumbnail", (
+            "the photo lost its thumbnail")
+        assert video.spawned == ["clip.mkv"] and photo.spawned == ["shot.jpg"], (
+            "unchanged files were probed and thumbnailed all over again")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
diff --git a/packages/meshbay-node/tests/test_root_availability.py b/packages/meshbay-node/tests/test_root_availability.py
index 0201c1f..d514dee 100644
--- a/packages/meshbay-node/tests/test_root_availability.py
+++ b/packages/meshbay-node/tests/test_root_availability.py
@@ -26,9 +26,12 @@ from meshbay_node.roots import RootSet
 pytestmark = pytest.mark.asyncio
 
 
-def _roots(*paths: Path) -> RootSet:
+def _roots(*paths: Path, removable: bool = False) -> RootSet:
     specs = [{"path": str(p)} for p in paths]
-    specs[0]["upload"] = True
+    specs[0]["writable"] = True
+    if removable:
+        for spec in specs:
+            spec["removable"] = True
     return RootSet.build(specs)
 
 
@@ -117,7 +120,9 @@ async def test_members_are_told_which_roots_are_unavailable(tmp_path):
 
     idx = await _indexer(_roots(films))
     assert idx.index.roots == [
-        {"name": "Films", "kind": "generic", "available": True, "upload": True}]
+        {"name": "Films", "kind": "generic", "available": True,
+         "writable": True, "removable": False, "ejected": False,
+         "upload": True}]
 
     (films / "a.mkv").unlink()
     films.rmdir()
diff --git a/packages/meshbay-node/tests/test_root_eject.py b/packages/meshbay-node/tests/test_root_eject.py
new file mode 100644
index 0000000..f57fb92
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_eject.py
@@ -0,0 +1,267 @@
+"""
+Safe eject, and the surprise unplug it exists to survive.
+
+`test_root_availability.py` pins the freeze: a root that goes away keeps its
+entries. This pins the half the operator drives — telling the node the drive is
+about to leave, and telling it the drive is back.
+
+The distinction that makes any of this work is that `ejected` and `is_live()`
+are separate answers. Between clicking Eject and physically unplugging, the
+directory is still readable; a design that recomputed availability from the
+filesystem alone would flip the root straight back to available and start
+serving files from a disk somebody has their hand on.
+
+The other property here is that the flag is *persisted*. It reached the roster
+in the first implementation and was never read back, so a restart — which is
+exactly what an operator does after noticing a drive fell off — silently undid
+the eject, and the next scan read an empty mount point as an erased library.
+"""
+
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+
+def _roots(*paths: Path, removable: bool = True) -> RootSet:
+    return RootSet.build([
+        {"path": str(p), "removable": removable} for p in paths])
+
+
+async def _indexer(roots: RootSet, **kw) -> DirectoryIndexer:
+    idx = DirectoryIndexer(roots=roots, group_id="g" * 32,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None, **kw)
+    await idx.initial_scan()
+    return idx
+
+
+def _names(idx: DirectoryIndexer) -> set[str]:
+    return {e.name for e in idx.index.entries}
+
+
+# ── The two states are not the same question ─────────────────────────────────
+
+async def test_ejecting_hides_a_root_that_is_still_readable(tmp_path):
+    """
+    The whole point of an eject button: the operator says the drive is leaving
+    *before* it leaves. The directory is still there and still readable at this
+    moment, so anything deriving availability from the filesystem would refuse
+    to believe it.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+
+    assert films.is_dir(), "the drive has not been unplugged yet"
+    assert roots.roots[0].is_live() is True
+    assert roots.roots[0].available is False
+    assert idx.index.roots[0]["ejected"] is True
+    assert idx.index.roots[0]["available"] is False
+
+
+async def test_an_eject_freezes_entries_rather_than_dropping_them(tmp_path):
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    (films / "b.mkv").write_bytes(b"b")
+
+    idx = await _indexer(_roots(films))
+    idx.eject_root("Films")
+
+    assert _names(idx) == {"a.mkv", "b.mkv"}, "eject deleted entries"
+
+
+async def test_reconciling_does_not_un_eject_a_root(tmp_path):
+    """
+    The backstop runs every minute regardless. An ejected root whose directory
+    is still readable must stay ejected, or the operator's eject lasts until
+    the next tick.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+    await idx.reconcile()
+
+    assert roots.roots[0].ejected is True
+    assert roots.roots[0].available is False
+
+
+async def test_plugging_back_relists_the_files(tmp_path):
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+    await idx.plug_root("Films")
+
+    assert roots.roots[0].ejected is False
+    assert roots.roots[0].available is True
+    assert _names(idx) == {"a.mkv"}
+
+
+async def test_what_changed_while_unplugged_is_picked_up_on_plug(tmp_path):
+    """
+    A drive people take away comes back different. The plug pass has to see
+    that, or the index describes a library that no longer exists on the disk
+    the node is about to serve from.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+
+    (films / "a.mkv").unlink()
+    (films / "c.mkv").write_bytes(b"c")
+
+    await idx.plug_root("Films")
+    assert _names(idx) == {"c.mkv"}
+
+
+# ── The surprise unplug ──────────────────────────────────────────────────────
+
+async def test_a_removable_root_that_vanishes_is_auto_ejected(tmp_path):
+    """
+    Nobody clicks Eject when they are in a hurry. A removable root whose path
+    disappears is treated as ejected rather than merely unavailable, so it does
+    not silently come back the moment the same mount point is readable again —
+    which on a machine with automount is any other drive, or an empty stub.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+
+    assert roots.roots[0].ejected is True
+    assert _names(idx) == {"a.mkv"}, "the library was treated as erased"
+
+
+async def test_a_non_removable_root_is_not_auto_ejected(tmp_path):
+    """
+    The counter-property. Auto-eject requires the operator to have said the
+    device is removable; an ordinary directory that briefly fails to stat must
+    keep the old behaviour and come back on its own.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films, removable=False)
+    idx = await _indexer(roots)
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+    assert roots.roots[0].ejected is False
+    assert roots.roots[0].available is False
+
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    await idx.reconcile()
+    assert roots.roots[0].available is True
+
+
+async def test_an_auto_eject_is_reported_so_it_can_be_persisted(tmp_path):
+    """
+    The flag has to outlive the process. The first version of this set it in
+    memory only, so restarting the node — which is what an operator does after
+    noticing a drive fell off — cleared it, and the scan that followed read the
+    empty mount point as a deletion of the whole library.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    seen: list[tuple[str, bool]] = []
+
+    async def record(name: str, ejected: bool) -> None:
+        seen.append((name, ejected))
+
+    roots = _roots(films)
+    idx = await _indexer(roots, on_root_ejected=record)
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+
+    assert seen == [("Films", True)]
+
+    # And only once, however many times the backstop runs afterwards.
+    await idx.reconcile()
+    await idx.reconcile()
+    assert seen == [("Films", True)]
+
+
+# ── Restoring the flag ───────────────────────────────────────────────────────
+
+async def test_a_root_built_as_ejected_starts_unavailable(tmp_path):
+    """
+    What the daemon does with what the roster remembers. `available` must not
+    be left at its default `True` here, or the group serves a drive that is not
+    there for as long as it takes the first reconcile to run.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    roots = RootSet.build([{"path": str(films), "removable": True,
+                            "ejected": True}])
+    assert roots.roots[0].ejected is True
+    assert roots.roots[0].available is False
+
+
+async def test_the_roster_round_trips_the_ejected_set(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.ejected_roots("g1") == set()
+
+        await roster.set_root_ejected("g1", "Films", True, set_by="op")
+        await roster.set_root_ejected("g1", "Music", False, set_by="op")
+        assert await roster.ejected_roots("g1") == {"films"}
+
+        # Another group's drives are its own.
+        assert await roster.ejected_roots("g2") == set()
+
+        await roster.set_root_ejected("g1", "Films", False, set_by="op")
+        assert await roster.ejected_roots("g1") == set()
+    finally:
+        await roster.close()
+
+
+async def test_the_ejected_key_is_case_folded(tmp_path):
+    """
+    Root names are compared without regard to case everywhere else, and a key
+    that did not fold would let `Films` and `films` disagree about the same
+    drive — on Windows and macOS, the same directory.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_root_ejected("g1", "FILMS", True, set_by="op")
+        assert await roster.ejected_roots("g1") == {"films"}
+        assert Roster.root_ejected_key("Films") == Roster.root_ejected_key("FILMS")
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py b/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py
new file mode 100644
index 0000000..976af82
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py
@@ -0,0 +1,361 @@
+"""
+Adding or removing a root has to reach the running node, not only node.toml.
+
+Two front doors do this — the loopback API and a signed MNP op — and `ops.py`
+exists so they behave identically. They did not. The loopback path fired the
+daemon's `reload_fn`, which re-reads node.toml and builds a fresh `RootSet`;
+the MNP path instead re-pointed the indexer at `groups_ctx[gid]["roots"]`, the
+very object the op had just been asked about. `DirectoryIndexer.retarget`
+decides what to scan by diffing the names it holds against the ones it is
+given, so a set compared against itself scans nothing and drops nothing.
+
+A directory added from a browser therefore reached node.toml and was invisible
+everywhere else until a restart — and adding it again was refused as colliding
+with itself, which is the only reason anyone found out. One removed would have
+kept serving its files.
+
+**The obvious repair is wrong in the other direction**, and was committed once
+before this file said so: making the op edit the live set in place puts the new
+root on *both* sides of retarget's comparison. The table would show it and it
+would stay permanently empty. So the ops leave that object alone, the MNP path
+reloads like the loopback one always did, and the tests below check the files —
+`describe()` agreeing proves nothing about whether anything was scanned.
+"""
+
+from dataclasses import asdict
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node import ops
+from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+async def _state(tmp_path: Path) -> tuple[dict, Roster]:
+    """A node hosting one group with two roots, as node.toml and as live state."""
+    for name in ("one", "two"):
+        (tmp_path / name).mkdir()
+
+    cfg = GroupConfig(id=GROUP, name="plop", roots=[
+        RootSpec(path=str(tmp_path / "one"), name="one"),
+        RootSpec(path=str(tmp_path / "two"), name="two"),
+    ])
+    node_cfg = NodeConfig.__new__(NodeConfig)
+    node_cfg.groups = [cfg]
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GROUP}"\nname = "plop"\n\n'
+        f'  [[groups.roots]]\n  path = "{(tmp_path / "one").as_posix()}"\n'
+        f'  name = "one"\n\n'
+        f'  [[groups.roots]]\n  path = "{(tmp_path / "two").as_posix()}"\n'
+        f'  name = "two"\n')
+
+    live = RootSet.build([asdict(r) for r in cfg.roots])
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state = {
+        "config": node_cfg,
+        "config_path": str(conf),
+        "groups_ctx": {GROUP: {"index": index, "roots": live}},
+        "roster": roster,
+        "node_user_id": "operator",
+    }
+    return state, roster
+
+
+def _live(state) -> RootSet:
+    return state["groups_ctx"][GROUP]["roots"]
+
+
+async def _indexer(state) -> DirectoryIndexer:
+    idx = DirectoryIndexer(roots=_live(state), group_id=GROUP,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    return idx
+
+
+def _rebuilt(state) -> RootSet:
+    """What a reload produces: a fresh set from the config the op just wrote."""
+    return RootSet.build([asdict(r) for r in state["config"].groups[0].roots])
+
+
+# ── What the op writes ───────────────────────────────────────────────────────
+
+async def test_adding_a_root_reaches_node_toml_and_the_ack(tmp_path):
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    try:
+        result = await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+
+        assert [r["name"] for r in result["roots"]] == ["one", "two", "uploads"]
+        assert [r.name for r in state["config"].groups[0].roots] == [
+            "one", "two", "uploads"]
+        assert "uploads" in Path(state["config_path"]).read_text()
+    finally:
+        await roster.close()
+
+
+async def test_removing_a_root_reaches_node_toml_and_the_ack(tmp_path):
+    state, roster = await _state(tmp_path)
+    try:
+        result = await ops.remove_root(state, GROUP, "two")
+        assert [r["name"] for r in result["roots"]] == ["one"]
+        assert Path(state["config_path"]).read_text().count(
+            "[[groups.roots]]") == 1
+    finally:
+        await roster.close()
+
+
+async def test_the_op_does_not_edit_the_live_set_in_place(tmp_path):
+    """
+    The property that made the original bug, and then made the first repair for
+    it wrong in the other direction.
+
+    `retarget` diffs the names it holds against the ones it is handed. Editing
+    that same object and passing it back puts a new root on both sides of the
+    comparison: nothing is scanned, and the directory shows in the table
+    permanently empty. `_reload_config_inner` diffs the same way and would
+    likewise conclude nothing had changed.
+    """
+    state, roster = await _state(tmp_path)
+    before = [r.name for r in _live(state)]
+    (tmp_path / "uploads").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        assert [r.name for r in _live(state)] == before, (
+            "add_root edited the live RootSet, which is the object retarget "
+            "diffs against — the new root would never be scanned")
+
+        await ops.remove_root(state, GROUP, "two")
+        assert [r.name for r in _live(state)] == before, (
+            "remove_root edited the live RootSet, so retarget cannot tell the "
+            "removed root's entries should go")
+    finally:
+        await roster.close()
+
+
+# ── What the node then serves ────────────────────────────────────────────────
+
+async def test_a_retarget_from_the_config_scans_the_new_root(tmp_path):
+    """
+    The half no assertion about `describe()` can reach: the files.
+
+    A root that appears in the table and holds nothing is the same bug one step
+    later, and it is what editing the live set in place would produce.
+    """
+    state, roster = await _state(tmp_path)
+    (tmp_path / "one" / "kept.txt").write_bytes(b"kept")
+    fresh = tmp_path / "uploads"
+    fresh.mkdir()
+    (fresh / "new.txt").write_bytes(b"new")
+
+    idx = await _indexer(state)
+    assert {e.name for e in idx.index.entries} == {"kept.txt"}
+    try:
+        await ops.add_root(state, GROUP, str(fresh))
+        await idx.retarget(_rebuilt(state))
+
+        assert {e.name for e in idx.index.entries} == {"kept.txt", "new.txt"}, (
+            "the added directory was not scanned — it would show in the table "
+            "and stay empty")
+        assert [r["name"] for r in idx.index.roots] == ["one", "two", "uploads"]
+    finally:
+        await roster.close()
+
+
+async def test_handing_retarget_the_edited_set_scans_nothing(tmp_path):
+    """
+    The failure mode above, demonstrated rather than described — so the reason
+    the ops leave the live set alone is checkable instead of asserted in a
+    comment. If this ever starts failing, `retarget` has changed and the rule
+    in `add_root` can be revisited.
+    """
+    state, roster = await _state(tmp_path)
+    fresh = tmp_path / "uploads"
+    fresh.mkdir()
+    (fresh / "new.txt").write_bytes(b"new")
+
+    idx = await _indexer(state)
+    try:
+        await ops.add_root(state, GROUP, str(fresh))
+        # What editing in place would have left behind.
+        _live(state).roots.append(_rebuilt(state).roots[-1])
+        await idx.retarget(_live(state))
+
+        assert {e.name for e in idx.index.entries} == set(), (
+            "retarget now scans a root it was handed on both sides of its own "
+            "diff — the constraint this file is built on has changed")
+    finally:
+        await roster.close()
+
+
+async def test_a_retarget_from_the_config_drops_a_removed_root(tmp_path):
+    """The mirror: a removed directory's files must stop being served."""
+    state, roster = await _state(tmp_path)
+    (tmp_path / "one" / "kept.txt").write_bytes(b"kept")
+    (tmp_path / "two" / "going.txt").write_bytes(b"going")
+
+    idx = await _indexer(state)
+    assert {e.name for e in idx.index.entries} == {"kept.txt", "going.txt"}
+    try:
+        await ops.remove_root(state, GROUP, "two")
+        await idx.retarget(_rebuilt(state))
+        assert {e.name for e in idx.index.entries} == {"kept.txt"}, (
+            "the removed directory's files are still being served")
+    finally:
+        await roster.close()
+
+
+# ── The invariants around them ───────────────────────────────────────────────
+
+async def test_adding_the_same_directory_twice_is_still_refused(tmp_path):
+    """A group with one path under two names indexes every file in it twice."""
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        with pytest.raises(ops.OpError):
+            await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        assert len(state["config"].groups[0].roots) == 3, (
+            "the refused add left something behind")
+        assert Path(state["config_path"]).read_text().count(
+            "[[groups.roots]]") == 3
+    finally:
+        await roster.close()
+
+
+async def test_a_second_different_root_still_lands(tmp_path):
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    (tmp_path / "incoming").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        result = await ops.add_root(state, GROUP, str(tmp_path / "incoming"),
+                                    writable=True)
+        assert [r["name"] for r in result["roots"]] == [
+            "one", "two", "uploads", "incoming"]
+        assert result["roots"][-1]["writable"] is True
+    finally:
+        await roster.close()
+
+
+async def test_updating_flags_may_edit_the_live_set(tmp_path):
+    """
+    The exception, and why it is one: `writable` and `removable` change nothing
+    about which files exist, so there is nothing for retarget to scan or drop.
+    Editing in place is what makes the flag true for the upload handler on the
+    very next request, which is synchronous and reads the live set.
+    """
+    state, roster = await _state(tmp_path)
+    state["config"] = SimpleNamespace(groups=state["config"].groups)
+    try:
+        result = await ops.update_root(state, GROUP, "two",
+                                       writable=True, removable=True)
+        live = _live(state).by_name("two")
+        assert live.writable is True and live.removable is True
+        assert result["roots"] == _live(state).describe()
+    finally:
+        await roster.close()
+
+
+async def test_the_file_on_disk_and_the_config_in_memory_agree(tmp_path):
+    """
+    A reload re-reads the file, so a config edited in memory but not on disk is
+    undone by the next restart — and one written to disk but not in memory
+    makes the *next* op validate against a stale picture.
+    """
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"),
+                           writable=True)
+        await ops.remove_root(state, GROUP, "one")
+
+        import tomllib
+        on_disk = tomllib.loads(Path(state["config_path"]).read_text())
+        disk_paths = [str(r["path"]) for r in on_disk["groups"][0]["roots"]]
+        memory_paths = [Path(r.path).as_posix()
+                        for r in state["config"].groups[0].roots]
+        assert disk_paths == memory_paths
+        assert Path(state["config_path"]).read_text().count(
+            "[[groups.roots]]") == 2
+    finally:
+        await roster.close()
+
+
+# ── The seam that was actually broken ────────────────────────────────────────
+
+async def test_the_mnp_path_reloads_like_the_loopback_one(tmp_path):
+    """
+    The two front doors, doing the same thing.
+
+    `ui/app.py` has always fired the daemon's `reload_fn` after a root op.
+    `_retarget_indexer` did not — it re-pointed the indexer at the live set
+    instead, which is the object the ops leave alone, so nothing happened at
+    all. That divergence *is* the bug: the loopback path worked, the MNP path
+    did not, and it survived until an operator added a directory from a
+    browser.
+
+    Not awaited: a reload rescans, and a new library is minutes. The ack
+    already carries the set the node is moving to.
+    """
+    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+    state, roster = await _state(tmp_path)
+    reloaded: list[bool] = []
+
+    async def fake_reload():
+        reloaded.append(True)
+
+    state["reload_fn"] = fake_reload
+    spawned = []
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"daemon_state": state}
+    session._spawn = lambda coro: spawned.append(coro)
+    try:
+        await session._retarget_indexer(GROUP)
+        assert spawned, "the MNP path did not ask the daemon to reload"
+        await spawned[0]
+        assert reloaded == [True]
+    finally:
+        await roster.close()
+
+
+async def test_without_a_daemon_it_still_retargets(tmp_path):
+    """
+    A context assembled by hand — a harness, or a test — has no `reload_fn`.
+    Falling through to a direct retarget keeps those working, and is correct
+    precisely because the ops no longer edit the set being passed.
+    """
+    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+    state, roster = await _state(tmp_path)
+    fresh = tmp_path / "uploads"
+    fresh.mkdir()
+    (fresh / "new.txt").write_bytes(b"new")
+    idx = await _indexer(state)
+    state["indexers"] = {GROUP: idx}
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"daemon_state": state}
+    try:
+        await ops.add_root(state, GROUP, str(fresh))
+        # What a reload would have installed, done by hand here.
+        state["groups_ctx"][GROUP]["roots"] = _rebuilt(state)
+        await session._retarget_indexer(GROUP)
+        assert {e.name for e in idx.index.entries} == {"new.txt"}
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_root_paths_are_operator_only.py b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py
new file mode 100644
index 0000000..080d4be
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py
@@ -0,0 +1,114 @@
+"""
+Where a directory lives on the operator's disk is theirs, not the group's.
+
+`RootSet.describe()` feeds two very different audiences. The index payload goes
+to every member, and has always deliberately carried no paths — a member is
+told what exists and whether it is readable, not that the library sits in
+`/media/<the operator's name>/BACKUP2`. The loopback API answers the operator
+themselves, over a channel that already requires being on their machine with
+the run token, where the path is exactly what they are asking for.
+
+`meshbay-node root list` printed `?` for every directory because it read a
+field the member form omits. Nothing caught it: the CLI reads a dict, the
+payload is a dict, and neither end says what keys it owes the other.
+
+Both halves matter and they pull opposite ways, so both are asserted here — a
+test that only checked the operator gets paths would be satisfied by putting
+them in the member payload too.
+"""
+
+import inspect
+import re
+from pathlib import Path
+
+
+from meshbay_node import daemon as daemon_mod
+from meshbay_node import ops
+from meshbay_node.roots import RootSet
+
+
+def _roots(tmp_path: Path) -> RootSet:
+    for name in ("Films", "Albums"):
+        (tmp_path / name).mkdir()
+    return RootSet.build([
+        {"path": str(tmp_path / "Films"), "writable": True},
+        {"path": str(tmp_path / "Albums"), "removable": True},
+    ])
+
+
+# ── The member's half ────────────────────────────────────────────────────────
+
+def test_the_default_form_carries_no_path(tmp_path):
+    described = _roots(tmp_path).describe()
+    assert described, "no roots described"
+    assert not any("path" in d for d in described), (
+        "the index payload every member receives would carry the operator's "
+        "filesystem layout")
+
+
+def test_the_default_form_still_says_what_a_member_needs(tmp_path):
+    """The counter-property: dropping the path must not drop the rest."""
+    described = _roots(tmp_path).describe()
+    for d in described:
+        assert set(d) >= {"name", "kind", "available", "writable",
+                          "removable", "ejected"}
+
+
+def test_the_index_payload_is_built_without_paths():
+    """
+    Read from the source, because the alternative is asserting it about a
+    payload built by a test rather than by the node.
+    """
+    from meshbay_node.indexer import indexer as indexer_mod
+    source = inspect.getsource(indexer_mod)
+    for call in re.findall(r"roots\.describe\([^)]*\)", source):
+        assert "with_paths" not in call, (
+            f"the indexer builds the member-facing roots table as {call} — "
+            f"that payload goes to everyone in the group")
+
+
+# ── The operator's half ──────────────────────────────────────────────────────
+
+def test_the_operator_form_carries_the_path(tmp_path):
+    described = _roots(tmp_path).describe(with_paths=True)
+    assert all(d.get("path") for d in described)
+    assert described[0]["path"] == str(tmp_path / "Films")
+
+
+def test_the_loopback_api_asks_for_paths():
+    """
+    `list_groups` answers the operator's own channel, and the CLI's `root list`
+    prints what it returns. Asking for the member form there is what printed a
+    column of question marks.
+    """
+    source = inspect.getsource(ops.list_groups)
+    assert "describe(with_paths=True)" in source, (
+        "list_groups uses the member form, so every path it reports is missing")
+
+
+def test_the_cli_only_reads_fields_the_payload_carries():
+    """
+    The gap this whole file exists for. The CLI reads a dict and the API
+    returns a dict; nothing between them says which keys are owed, so a name
+    that is simply absent prints as a placeholder and looks like a node
+    problem.
+    """
+    source = inspect.getsource(daemon_mod.main)
+    start = source.index('if args.command == "root":')
+    block = source[start:source.index('if args.command == "operator":', start)]
+
+    read = set(re.findall(r"r\.get\(['\"](\w+)['\"]", block))
+    read |= set(re.findall(r"r\[['\"](\w+)['\"]\]", block))
+    assert read, "the root CLI no longer reads the payload this way"
+
+    class _Any:
+        path = Path("/tmp/x")
+        name = "x"
+        kind = "generic"
+        writable = removable = ejected = False
+        available = True
+
+    offered = set(RootSet(roots=[_Any()]).describe(with_paths=True)[0])
+    assert read <= offered, (
+        f"the `root` CLI reads keys the loopback payload does not carry: "
+        f"{sorted(read - offered)}")
diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py
new file mode 100644
index 0000000..7eb75fd
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_writable_policy.py
@@ -0,0 +1,246 @@
+"""
+Who may write to the operator's disk, now that RO/RW on the root decides it.
+
+This replaces `test_member_upload_policy.py`. The old model had two orthogonal
+controls — one root designated as the upload target, and a group-wide
+`member_upload` switch — and collapsed into one property per root: `writable`.
+The properties worth keeping from the old file survive the change unaltered:
+
+* the interface hiding a control is a courtesy to the people who are not
+  trying; **the node refusing is the part that holds** against someone who is.
+  A member with an old tab open, or one speaking MNP directly, gets the same
+  answer. That half is pinned in `test_security_regressions.py`, next to the
+  overwrite properties it belongs with;
+* the setting is changed by a **signed** operator instruction, or it is a
+  suggestion any member can undo;
+* it is stored on the **node**, never the hub. A hub that could decide who
+  writes to the operator's disk would have authority over the node.
+
+And one that is new: the *old* message must no longer be able to change
+anything. A deprecated instruction that still works is not deprecated, and this
+one would reopen uploads group-wide.
+"""
+
+import base64
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *,
+             writable: bool = True,
+             operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": RootSet.build([{"path": str(shared_root), "writable": writable}]),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = "g" * 32
+    session._user_id = user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _upload(session, filename="clip.mp4", body=b"bytes"):
+    session._do_file_upload({
+        "filename": filename, "dir": "shared",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(body).decode(),
+    })
+
+
+def _uploads_dir(session) -> Path:
+    # The root itself: the `uploads/` subdirectory the node used to create is
+    # gone (see test_security_regressions._uploads_dir for why).
+    return session._ctx["roots"].roots[0].path
+
+
+# ── The door, not the button ─────────────────────────────────────────────────
+
+async def test_a_member_cannot_upload_to_a_read_only_root(tmp_path):
+    session = _session(tmp_path, "member-1", writable=False)
+    _upload(session)
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not (_uploads_dir(session) / "clip.mp4").exists()
+
+
+async def test_members_upload_normally_to_a_writable_root(tmp_path):
+    session = _session(tmp_path, "member-1", writable=True)
+    _upload(session)
+
+    assert not [m for m in session.sent if m.get("type") == "error"]
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_read_only_binds_the_operator_too(tmp_path):
+    """
+    The old model exempted the operator, because the switch was about *members*.
+    RO is about the directory: a published library is read-only for everyone, and
+    an exception for admin authority is how a rule turns into a default.
+    """
+    session = _session(tmp_path, "the-operator", writable=False,
+                       operator="the-operator")
+    session._is_node_admin = lambda: True
+    _upload(session)
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+
+
+async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path):
+    """
+    Read-only has to mean read-only for every way of writing, not just for
+    files. `_do_file_upload` gained this check with the RO/RW model and
+    `_do_dir_create` did not, so a member refused a file in a published library
+    could still leave empty directories all through it.
+
+    Creating a folder stays unprivileged — the node's own words: "a member who
+    can add a file can organise where it goes". What changed is that it now
+    requires the same root to be writable that adding the file would have.
+    """
+    session = _session(tmp_path, "member-1", writable=False)
+    session._do_dir_create({"dir": "shared", "name": "New folder"})
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not (tmp_path / "shared" / "New folder").exists()
+
+
+async def test_a_member_can_create_a_folder_in_a_writable_root(tmp_path):
+    """The counter-property: it must stay unprivileged where it is allowed."""
+    session = _session(tmp_path, "member-1", writable=True)
+    session._do_dir_create({"dir": "shared", "name": "New folder"})
+
+    assert not [m for m in session.sent if m.get("type") == "error"]
+    assert (tmp_path / "shared" / "New folder").is_dir()
+
+
+async def test_an_ejected_root_refuses_a_new_folder(tmp_path):
+    """Writing to a drive somebody has their hand on, one level up from a file."""
+    session = _session(tmp_path, "member-1", writable=True)
+    roots = session._ctx["roots"]
+    roots.roots[0].ejected = True
+    roots.roots[0].available = False
+
+    session._do_dir_create({"dir": "shared", "name": "New folder"})
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_unavailable"
+    assert not (tmp_path / "shared" / "New folder").exists()
+
+
+# ── Signed, or it is a suggestion ────────────────────────────────────────────
+
+def _capture_challenges(session) -> list[tuple[str, str]]:
+    issued: list[tuple[str, str]] = []
+
+    def issue(op, subject, **kw):
+        issued.append((op, subject))
+
+    session._issue_admin_challenge = issue
+    session._has_admin_authority = lambda: True
+    return issued
+
+
+async def test_changing_a_roots_flags_needs_a_signature(tmp_path):
+    """The flags are not applied by the request — only by the signed response."""
+    session = _session(tmp_path, "the-operator", operator="the-operator")
+    issued = _capture_challenges(session)
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": False})
+
+    assert [op for op, _ in issued] == [OP_ROOT_UPDATE]
+    assert session._ctx["roots"].roots[0].writable is True, (
+        "applied before it was signed")
+
+
+async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
+    """
+    The operator is shown the subject before signing, so it has to say what will
+    be true afterwards. "shared" alone would have them authorize a change they
+    cannot see the direction of.
+    """
+    session = _session(tmp_path, "op", operator="op")
+    issued = _capture_challenges(session)
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": True, "removable": True})
+
+    assert issued == [(OP_ROOT_UPDATE, "shared:rw=on,rem=on")]
+
+
+async def test_eject_and_plug_are_signed_too(tmp_path):
+    """
+    Hiding a group's whole library from every member is not a lesser act than
+    changing a flag. An unsigned one would let any member black out a group.
+    """
+    session = _session(tmp_path, "op", operator="op")
+    issued = _capture_challenges(session)
+
+    session._do_root_eject({"group_id": "g" * 32, "root_name": "shared"})
+    session._do_root_plug({"group_id": "g" * 32, "root_name": "shared"})
+
+    assert issued == [(OP_ROOT_EJECT, "shared"), (OP_ROOT_PLUG, "shared")]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    """
+    An unpaired node has no key to check a signature against, so the challenge
+    is never issued rather than issued and then unverifiable.
+    """
+    session = _session(tmp_path, "member-1")
+    issued = _capture_challenges(session)
+    session._has_admin_authority = lambda: False
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": True})
+
+    assert issued == []
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── The deprecated message must not still work ───────────────────────────────
+
+async def test_the_old_member_upload_message_changes_nothing(tmp_path):
+    """
+    MNP still parses `member_upload` so an old client gets an answer instead of
+    a dropped request. What it must not do is act: this instruction could
+    reopen uploads for a whole group, and a client old enough to send it is
+    exactly one that knows nothing about read-only roots.
+    """
+    session = _session(tmp_path, "member-1", writable=False)
+    session._has_admin_authority = lambda: True
+    issued = _capture_challenges(session)
+
+    session._do_member_upload({"allowed": True})
+
+    assert issued == [], "a deprecated instruction asked to be signed"
+    assert session._ctx["roots"].roots[0].writable is False
+    acks = [m for m in session.sent if m.get("type") == MNP.MEMBER_UPLOAD_ACK]
+    assert acks and acks[0].get("deprecated") is True
+
+    # And the door is still shut.
+    _upload(session)
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py
index 1beb220..505091b 100644
--- a/packages/meshbay-node/tests/test_roots.py
+++ b/packages/meshbay-node/tests/test_roots.py
@@ -108,31 +108,61 @@ def test_a_sibling_with_a_shared_prefix_is_fine(tmp_path):
     assert roots.names == ["Media", "Media2"]
 
 
-# ── Uploads ──────────────────────────────────────────────────────────────────
+# ── Writable roots ───────────────────────────────────────────────────────────
 
-def test_a_single_root_receives_uploads_without_being_asked(tmp_path):
+def test_a_root_is_read_only_unless_it_says_otherwise(tmp_path):
+    """
+    The default is the safe one. An operator who shares a directory has not
+    thereby agreed to let anyone write into it, and the version of this that
+    guessed — one root, so it must be the upload target — meant adding a
+    second directory silently changed what the first one was.
+    """
     (tmp_path / "Media").mkdir()
     roots = RootSet.build([_spec(tmp_path / "Media")])
-    assert roots.upload_root is roots.roots[0]
+    assert roots.roots[0].writable is False
+    assert roots.writable_roots == []
+
+
+def test_several_roots_can_be_writable_at_once(tmp_path):
+    (tmp_path / "A").mkdir()
+    (tmp_path / "B").mkdir()
+    (tmp_path / "C").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "A", writable=True),
+                           _spec(tmp_path / "B"),
+                           _spec(tmp_path / "C", writable=True)])
+    assert [r.name for r in roots.writable_roots] == ["A", "C"]
 
 
-def test_several_roots_and_no_designation_means_no_uploads(tmp_path):
+def test_a_fully_read_only_group_is_valid(tmp_path):
     """
-    Refused, never guessed: picking one would send a member's file to a disk the
-    operator did not intend, and that is discovered weeks later.
+    A group that only publishes is the point of the read-only model, not a
+    misconfiguration — build must not refuse it, and nothing downstream may
+    promote a root to writable to have somewhere to put an upload.
     """
     (tmp_path / "A").mkdir()
     (tmp_path / "B").mkdir()
     roots = RootSet.build([_spec(tmp_path / "A"), _spec(tmp_path / "B")])
-    assert roots.upload_root is None
+    assert roots.writable_roots == []
+    assert len(roots) == 2
 
 
-def test_two_upload_roots_are_refused(tmp_path):
-    (tmp_path / "A").mkdir()
-    (tmp_path / "B").mkdir()
-    with pytest.raises(RootError, match="exactly one"):
-        RootSet.build([_spec(tmp_path / "A", upload=True),
-                       _spec(tmp_path / "B", upload=True)])
+def test_the_old_upload_flag_still_reads_as_writable(tmp_path):
+    """A node.toml written before this refactor must not change meaning."""
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", upload=True)])
+    assert roots.roots[0].writable is True
+    assert roots.describe()[0]["writable"] is True
+
+
+def test_writable_wins_over_a_leftover_upload_flag(tmp_path):
+    """
+    A config carrying both is one a migration touched. `writable` is the field
+    the operator's tooling writes now, so it is the one that decides — reading
+    the legacy field there would undo the migration on the next load.
+    """
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", upload=True, writable=False)])
+    assert roots.roots[0].writable is False
 
 
 # ── Resolution ───────────────────────────────────────────────────────────────
@@ -236,18 +266,35 @@ def test_availability_follows_the_directory(tmp_path):
 def test_describe_reports_what_a_member_needs(tmp_path):
     (tmp_path / "Media").mkdir()
     (tmp_path / "Music").mkdir()
-    roots = RootSet.build([_spec(tmp_path / "Media", upload=True),
-                           _spec(tmp_path / "Music", kind="audio")])
+    roots = RootSet.build([_spec(tmp_path / "Media", writable=True),
+                           _spec(tmp_path / "Music", kind="audio",
+                                 removable=True)])
     described = roots.describe()
     assert described == [
-        {"name": "Media", "kind": "generic", "available": True, "upload": True},
-        {"name": "Music", "kind": "audio", "available": True, "upload": False},
+        {"name": "Media", "kind": "generic", "available": True,
+         "writable": True, "removable": False, "ejected": False,
+         "upload": True},
+        {"name": "Music", "kind": "audio", "available": True,
+         "writable": False, "removable": True, "ejected": False,
+         "upload": False},
     ]
     # Deliberately no paths: a member is told what exists and whether it is
     # readable, not where on the operator's disk it lives.
     assert not any("path" in d for d in described)
 
 
+def test_describe_still_carries_upload_for_mnp_1_0_clients(tmp_path):
+    """
+    `upload` is `writable` under its old name, kept because an MNP 1.0 client
+    reads no other field and would otherwise decide the group takes no uploads
+    at all. It is derived, never stored — the two can never disagree.
+    """
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", writable=True)])
+    described = roots.describe()[0]
+    assert described["upload"] == described["writable"] is True
+
+
 # ── SAFE_UPLOAD_NAME ────────────────────────────────────────────────────────
 
 def test_safe_name_accepts_unicode_letters():
diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py
index 719b988..94f4421 100644
--- a/packages/meshbay-node/tests/test_scan_settings_policy.py
+++ b/packages/meshbay-node/tests/test_scan_settings_policy.py
@@ -2,7 +2,7 @@
 The operator can tune how often the indexer's reconciliation backstop runs,
 and how long it waits after a file's last write before hashing it.
 
-Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py:
+Same shape as test_apps_enabled_policy.py / test_root_writable_policy.py:
 changed by a signed operator instruction, stored on the node rather than the
 hub. Unlike those two, there is also a *live* DirectoryIndexer object to
 update — see test_set_scan_settings_updates_the_live_indexer below.
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 7f71da5..1a318f7 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -18,6 +18,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
 from conftest import one_root
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
@@ -132,14 +133,22 @@ def test_the_node_never_generates_a_name_it_would_refuse(tmp_path):
 
 def _uploads_dir(session) -> Path:
     """
-    Where this session's uploads land: uploads/ inside the group's upload root.
+    Where an unaddressed upload lands: the first writable root itself.
+
+    There is no `uploads/` subdirectory any more. It was the last of v5's
+    quarantine — the per-user layer went on 2026-08-14 — and it went for the
+    same reason: a folder appearing beside the operator's library because
+    somebody sent a file is the node deciding how their disk is arranged. The
+    protections that made the quarantine worth having are the allowlist, the
+    size cap, the chunk ordering and the no-overwrite rule, and every one of
+    them is asserted below, unchanged.
 
     Asked of the root set rather than assembled by hand, so a test cannot pass
     while agreeing with a wrong answer the code also produced.
     """
-    root = session._ctx["roots"].upload_root
-    assert root is not None, "the fixture must designate an upload root"
-    return root.path / "uploads"
+    writable = session._ctx["roots"].writable_roots
+    assert writable, "the fixture must give the group a writable root"
+    return writable[0].path
 
 
 def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
@@ -175,7 +184,6 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
     """
     victim = _session(tmp_path, "victim-user")
     uploads = _uploads_dir(victim)
-    uploads.mkdir()
     original = uploads / "important.mp4"
     original.write_bytes(b"operator's original content")
 
@@ -225,22 +233,157 @@ def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
     assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}"
 
 
-def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
+def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
+    """
+    The destination is now the folder the sender is looking at, which means the
+    client does choose it — and the whole of what keeps that safe is that the
+    choice is *resolved against the group's own roots* rather than joined to
+    one.
+
+    `RootSet.resolve()` refuses `..`, absolute segments and anything whose
+    resolved form escapes its root, symlinks included. So "which of this
+    group's folders" is answerable by a member and "which path on the
+    operator's disk" is not.
+    """
+    session = _session(tmp_path, "user-1")
+    (session._ctx["roots"].roots[0].path / "sub").mkdir()
+    before = set(tmp_path.rglob("*"))
+
+    for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc",
+                "nope", "shared/missing"):
+        session.sent.clear()
+        session._do_file_upload({
+            "filename": "note.txt", "dir": bad,
+            "chunk_index": 0, "total_chunks": 1,
+            "data": base64.b64encode(b"x").decode(),
+        })
+        refusal = [m for m in session.sent if m.get("type") == "error"]
+        assert refusal, f"{bad!r} was accepted"
+        assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad
+
+    assert set(tmp_path.rglob("*")) == before, "a refused upload still wrote"
+
+
+def test_an_upload_lands_in_the_folder_it_names(tmp_path):
+    """
+    And in that folder itself — the `uploads/` subdirectory the node used to
+    create is gone. Somebody dropping a file into the folder they are looking
+    at expects it to be in that folder.
+    """
+    session = _session(tmp_path, "user-1")
+    root = session._ctx["roots"].roots[0]
+    (root.path / "Albums").mkdir()
+
+    session._do_file_upload({
+        "filename": "note.txt", "dir": f"{root.name}/Albums",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    assert (root.path / "Albums" / "note.txt").read_bytes() == b"x"
+    assert not (root.path / "Albums" / "uploads").exists(), (
+        "the node invented a subdirectory in the operator's library")
+    assert not (root.path / "uploads").exists()
+
+
+def test_an_upload_goes_to_the_root_it_names(tmp_path):
+    """
+    With two writable roots there is no defensible default, and the client is
+    the only party that knows which directory the person is looking at. The
+    node picking one meant a file uploaded from a folder on screen landed in a
+    different one — the same "uploads went somewhere else" the single upload
+    root was never allowed to guess about.
+    """
+    media = tmp_path / "Media"
+    incoming = tmp_path / "Incoming"
+    media.mkdir()
+    incoming.mkdir()
+    session = _session(tmp_path, "user-1")
+    session._ctx["roots"] = RootSet.build([
+        {"path": str(media), "writable": True},
+        {"path": str(incoming), "writable": True},
+    ])
+
+    session._do_file_upload({
+        "filename": "note.txt", "dir": "Incoming",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    assert (incoming / "note.txt").read_bytes() == b"x"
+    assert not (media / "note.txt").exists(), "it went to the first root instead"
+
+
+def test_a_read_only_root_refuses_an_upload(tmp_path):
+    """
+    RO is the mechanism now, not a hidden button. It binds the operator too:
+    "read-only for everyone" is what makes a published library one, and an
+    exception for whoever happens to hold admin authority is the sort of
+    carve-out that later reads as the rule.
+    """
+    published = tmp_path / "Published"
+    published.mkdir()
+    session = _session(tmp_path, "user-1")
+    session._ctx["roots"] = RootSet.build([{"path": str(published)}])
+    session._is_node_admin = lambda: True
+
+    session._do_file_upload({
+        "filename": "note.txt", "dir": "Published",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not (published / "note.txt").exists()
+
+
+def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
+    """
+    An MNP 1.0 client names no root, so the node falls back to the first
+    writable one. There isn't one here, and the fallback must refuse rather
+    than write into whatever root happens to come first.
+    """
+    published = tmp_path / "Published"
+    published.mkdir()
+    session = _session(tmp_path, "user-1")
+    session._ctx["roots"] = RootSet.build([{"path": str(published)}])
+
+    session._do_file_upload({
+        "filename": "note.txt",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "no_writable_root"
+    assert not (published / "note.txt").exists()
+
+
+def test_an_ejected_root_refuses_an_upload(tmp_path):
     """
-    Uploads land in uploads/, chosen by the node. A client that names somewhere
-    else — or nowhere at all — changes nothing, so the traversal surface that a
-    client-chosen destination would open does not exist on this path.
+    Writing to a drive somebody has their hand on is the thing eject exists to
+    stop. `writable` is still true — that is configuration — so availability
+    has to be checked separately, which is what an earlier version conflated.
     """
+    usb = tmp_path / "USB"
+    usb.mkdir()
     session = _session(tmp_path, "user-1")
+    roots = RootSet.build([{"path": str(usb), "writable": True,
+                            "removable": True}])
+    roots.roots[0].ejected = True
+    roots.roots[0].available = False
+    session._ctx["roots"] = roots
 
     session._do_file_upload({
-        "filename": "note.txt", "dir": "../../etc",
+        "filename": "note.txt", "dir": "USB",
         "chunk_index": 0, "total_chunks": 1,
         "data": base64.b64encode(b"x").decode(),
     })
 
-    assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x"
-    assert not (tmp_path / "etc").exists()
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_unavailable"
+    assert not (usb / "note.txt").exists()
 
 
 def test_two_members_can_send_the_same_filename(tmp_path):
diff --git a/packages/meshbay-node/tests/test_startup_scan_enrichment.py b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
index 65b9728..d2cbc3e 100644
--- a/packages/meshbay-node/tests/test_startup_scan_enrichment.py
+++ b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
@@ -60,8 +60,9 @@ async def test_a_file_already_on_disk_at_startup_gets_enrichment_scheduled(tmp_p
         data_dir=tmp_path / "data",
     )
     class _StubRoster:
-        async def video_root(self, group_id):
-            return "shared"  # the root itself, i.e. "enrich the whole thing"
+        async def app_directories(self, group_id, app_key):
+            # The root itself, i.e. "enrich the whole thing".
+            return ["shared"] if app_key == "video" else []
 
     daemon = NodeDaemon(config)
     daemon._broadcast_coalesce_secs = 0.01  # real value would make this test wait 0.5s
diff --git a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
index 9cfb819..b88ecaf 100644
--- a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
+++ b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
@@ -99,7 +99,7 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
 
     daemon = await _make_daemon(tmp_path, shared, group_id)
     try:
-        await daemon._roster.set_video_root(group_id, "shared/Movies", set_by="op")
+        await daemon._roster.set_app_directories(group_id, "video", ["shared/Movies"], set_by="op")
         indexer = DirectoryIndexer(
             roots=one_root(shared), group_id=group_id,
             sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
@@ -139,8 +139,10 @@ async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path):
         # this file all along.
         state = {
             "roster": daemon._roster,
-            "groups_ctx": {group_id: {}},
-            "enrich_video_root_fn": daemon._enrich_video_root_now,
+            # Real roots, because ops now refuses a directory that is not
+            # inside one — the per-app setters this replaced validated nothing.
+            "groups_ctx": {group_id: {"roots": one_root(shared)}},
+            "enrich_app_dirs_fns": {"video": daemon._enrich_video_root_now},
         }
         await ops.set_video_root(state, group_id, "shared/Movies")
         await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
diff --git a/packages/meshbay-node/tests/test_video_root_policy.py b/packages/meshbay-node/tests/test_video_root_policy.py
index 8cc1540..8d8c45a 100644
--- a/packages/meshbay-node/tests/test_video_root_policy.py
+++ b/packages/meshbay-node/tests/test_video_root_policy.py
@@ -126,16 +126,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        assert await roster.video_root("g1") == "", "absent must mean the whole group index"
-        await roster.set_video_root("g1", "shared/Movies", set_by="op")
-        assert await roster.video_root("g1") == "shared/Movies"
+        assert await roster.app_directories("g1", "video") == [], (
+            "absent must mean nothing configured")
+        await roster.set_app_directories("g1", "video", ["shared/Movies"],
+                                         set_by="op")
+        assert await roster.app_directories("g1", "video") == ["shared/Movies"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.video_root("g1") == "shared/Movies"
-        assert await reopened.video_root("g2") == "", "one group's setting must not answer for another"
+        assert await reopened.app_directories("g1", "video") == ["shared/Movies"]
+        assert await reopened.app_directories("g2", "video") == [], (
+            "one group's setting must not answer for another")
     finally:
         await reopened.close()
diff --git a/packages/meshbay-node/tests/test_windows_root_shapes.py b/packages/meshbay-node/tests/test_windows_root_shapes.py
new file mode 100644
index 0000000..5c5da25
--- /dev/null
+++ b/packages/meshbay-node/tests/test_windows_root_shapes.py
@@ -0,0 +1,149 @@
+"""
+The root model against the shapes Windows produces.
+
+CLAUDE.md is explicit that exFAT/NTFS and Windows are the common case, not an
+edge case: most operators are expected to share from an external drive on
+Windows. The RO/RW refactor added two booleans and a config rewriter, and the
+booleans are path-independent — but the rewriter, the name derivation and the
+collision check all touch paths, and none of them has ever run on Windows here.
+
+What this can check without Windows is the *shape* work: drive letters through
+`as_posix()`, a path with no basename to derive a name from, UNC, and a
+case-insensitive collision. `PureWindowsPath` is used deliberately — the plain
+`Path` on this machine is a `PosixPath`, where a backslash is an ordinary
+filename character, which is the mistake that made
+`test_a_backslash_path_written_into_node_toml_stays_parseable` fail everywhere
+but the platform it was written for.
+
+What it cannot check is the filesystem itself: `ReadDirectoryChangesW` dropping
+events under load, `MAX_PATH`, and whether an eject actually lets a drive be
+removed. Those need a person with Windows, and §4.4 of the refactor plan is
+where that is written down.
+"""
+
+import os
+import tempfile
+import tomllib
+from pathlib import Path, PureWindowsPath
+
+import pytest
+
+from meshbay_node.roots import RootError, RootSet, derive_name
+
+BS = chr(92)
+
+
+# ── Paths into node.toml ─────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("raw,expected", [
+    (f"D:{BS}Movies", "D:/Movies"),
+    (f"E:{BS}Music{BS}Albums", "E:/Music/Albums"),
+    (f"C:{BS}Users{BS}alice{BS}Media", "C:/Users/alice/Media"),
+    (f"{BS}{BS}server{BS}share{BS}Media", "//server/share/Media"),
+])
+def test_a_windows_path_survives_the_config_file(raw, expected):
+    """
+    `ops` writes `as_posix()` into a TOML basic string, where a raw backslash
+    is an escape — `\\U` and `\\a` are the ones that bite — so the file would
+    not parse at all. pathlib reads the forward-slash form back on Windows.
+    """
+    posix = PureWindowsPath(raw).as_posix()
+    assert posix == expected
+    parsed = tomllib.loads(f'path = "{posix}"\n')
+    assert parsed["path"] == expected
+
+
+def test_a_raw_windows_path_would_not_parse():
+    """The counter-property: without `as_posix()` there is no config file."""
+    with pytest.raises(tomllib.TOMLDecodeError):
+        tomllib.loads(f'path = "C:{BS}Users{BS}alice{BS}Media"\n')
+
+
+# ── Naming a drive ───────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("raw,name", [
+    (f"D:{BS}Movies", "Movies"),
+    (f"E:{BS}Music{BS}Albums", "Albums"),
+    (f"{BS}{BS}server{BS}share{BS}Media", "Media"),
+])
+def test_a_name_is_derived_from_the_last_segment(raw, name):
+    assert PureWindowsPath(raw).name == name
+
+
+@pytest.mark.parametrize("raw", [f"D:{BS}", f"E:{BS}", f"{BS}{BS}server{BS}share"])
+def test_a_drive_root_has_no_name_to_derive(raw):
+    """
+    Sharing a whole drive is an ordinary thing to do on Windows and there is
+    nothing to call it, so the operator has to say. Refused with that as the
+    message rather than named "" or "D:".
+    """
+    p = PureWindowsPath(raw)
+    if p.name:
+        pytest.skip(f"{raw!r} has a basename on this platform")
+    with pytest.raises(RootError, match="explicit"):
+        derive_name(p)
+
+
+def test_naming_it_explicitly_works():
+    with tempfile.TemporaryDirectory() as d:
+        roots = RootSet.build([{"path": d, "name": "Films"}])
+        assert roots.names == ["Films"]
+
+
+# ── Case, which Windows makes real ───────────────────────────────────────────
+
+def test_two_roots_differing_only_in_case_are_refused():
+    """
+    On NTFS and exFAT `Movies` and `MOVIES` are the same directory to the
+    filesystem and two roots to a case-sensitive comparison — which would index
+    one tree twice, and make deleting a file from one copy break the other.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "Movies"))
+        os.makedirs(os.path.join(d, "other"))
+        with pytest.raises(RootError, match="regard to case"):
+            RootSet.build([
+                {"path": os.path.join(d, "Movies")},
+                {"path": os.path.join(d, "other"), "name": "MOVIES"},
+            ])
+
+
+def test_a_root_is_found_by_name_without_regard_to_case():
+    """
+    What a client sends is what a person typed or a path it split, and on
+    Windows those disagree about case routinely.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "Movies"))
+        roots = RootSet.build([{"path": os.path.join(d, "Movies")}])
+        for spelling in ("Movies", "movies", "MOVIES", "MoViEs"):
+            assert roots.by_name(spelling) is not None, spelling
+
+
+# ── The two flags ────────────────────────────────────────────────────────────
+
+def test_the_flags_do_not_touch_paths():
+    """
+    `writable` and `removable` are booleans and stay booleans on every
+    platform. Stated as a test because it is the reason the rest of the
+    refactor needed no Windows work: what did need it is above.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "USB"))
+        roots = RootSet.build([{"path": os.path.join(d, "USB"),
+                                "writable": True, "removable": True}])
+        described = roots.describe()[0]
+        assert described["writable"] is True
+        assert described["removable"] is True
+        assert "path" not in described
+
+
+def test_an_ejected_removable_root_is_unavailable_wherever_it_runs():
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "USB"))
+        roots = RootSet.build([{"path": os.path.join(d, "USB"),
+                                "removable": True, "ejected": True}])
+        assert roots.roots[0].available is False
+        assert Path(roots.roots[0].path).is_dir(), (
+            "the directory is still there; `ejected` is the operator's answer, "
+            "not the filesystem's")