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

9 files changed, 1713 insertions, 437 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")