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

1 files changed, 367 insertions, 98 deletions

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")