feat: groups refactor Phase 1 — root RO/RW model + shared directories UI

Replace the upload boolean with per-root writable/removable/ejected flags.
Backend: new ops (update_root, eject_root, plug_root), MNP 1.1 protocol
messages, live RootSet updates so API always reflects current state, CLI
root subcommand (add/remove/set/list/eject/plug).
Frontend: SharedDirectoriesTable with optimistic toggle switches, eject/plug
in Files and Settings, upload gated on root.writable, ejected-root filtering
in all media apps, updated Create Group wizard, 10-locale i18n.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

8 files changed, 745 insertions, 197 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index 0e5a7de..4712312 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,7 +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
+    writable:  bool = False   # RW roots accept uploads from group members
+    removable: bool = False   # operator can eject this root before unplugging the device
     direct: bool = False      # uploads land at root path, not in a subdirectory
 
 
@@ -201,7 +197,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 +220,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, direct=True))
 
 
 @dataclass
@@ -285,15 +281,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..f45101f 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -1663,15 +1663,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 +1689,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 "
@@ -1710,14 +1714,29 @@ def main() -> 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="mark root as read-write (root set/add)")
+    parser.add_argument("--no-writable", action="store_false",
+                        dest="writable",
+                        help="mark root as read-only (root set)")
+    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 +1980,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.
@@ -2408,11 +2434,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']}")
@@ -2450,10 +2483,18 @@ def main() -> None:
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
         body = {"name": args.target, "shared_dir": args.dir}
         if args.upload_dir:
+            import warnings
+            warnings.warn(
+                "--upload-dir is deprecated; the main root is writable by "
+                "default. Use 'meshbay-node root add' for additional roots.",
+                DeprecationWarning, stacklevel=1)
+            print("WARNING: --upload-dir is deprecated. The shared directory is "
+                  "writable by default. Use 'meshbay-node root add' for "
+                  "additional roots.")
             body["upload_dir"] = args.upload_dir
         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']}")
+        print(f"  shared_dir   {out['shared_dir']}  (writable)")
         if out.get("upload_dir"):
             print(f"  upload_dir   {out['upload_dir']}")
         print()
@@ -2465,6 +2506,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, f"/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..2911278 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -723,6 +723,44 @@ 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)
+            self._drop_root_entries(root)
+            await self._scan_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..c3a3f9c 100644
--- a/packages/meshbay-node/src/meshbay_node/ops.py
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -429,20 +429,16 @@ 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
+    # Legacy: upload_dir becomes a second writable root
     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:
@@ -451,9 +447,8 @@ async def attach_group(state: dict, name: str, shared_dir: str,
     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\n')
     try:
         with conf_path.open("a", encoding="utf-8", newline="\n") as f:
             f.write(block)
@@ -463,8 +458,6 @@ 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),
               "note": "restart the node to pick it up"}
-    if separate_upload:
-        result["upload_dir"] = str(upload_path)
     return result
 
 
@@ -641,7 +634,8 @@ def _remove_roots_block(conf_path: Path, group_id: str,
 
 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 +649,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,14 +669,17 @@ 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 += f'\n  writable = true'
+    if removable:
+        root_block += f'\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,
+        direct=added.direct))
 
     log.info("Root added: %s → group %s", added.name, group_id[:8])
     return {"status": "added", "name": added.name, "path": str(added.path),
@@ -714,25 +712,242 @@ 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
+
+    # Update the live RootSet so GET /api/groups returns correct data
+    # immediately, without waiting for the async reload.
+    live_roots = state.get("groups_ctx", {}).get(
+        group_id, {}).get("roots")
+    if live_roots:
+        live_roots.roots = [
+            r for r in live_roots.roots if fold(r.name) != target]
+
+    result_roots = live_roots.describe() if live_roots else []
 
     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
+
+    result_roots = live_roots.describe() if live_roots else []
+
+    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()}
+
+    root.ejected = True
+    root.available = False
+
+    roster = _roster(state)
+    if roster:
+        await roster.set_setting(
+            group_id, f"root_ejected:{fold(root_name)}", "1",
+            set_by=state.get("node_user_id", ""))
+
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        indexer.eject_root(root_name)
+
+    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)
+
+    root.ejected = False
+    root.available = True
+
+    roster = _roster(state)
+    if roster:
+        await roster.set_setting(
+            group_id, f"root_ejected:{fold(root_name)}", "0",
+            set_by=state.get("node_user_id", ""))
+
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        await indexer.plug_root(root_name)
+
+    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(f"Root path not found in config", status=404)
 
 
 # ── Files ────────────────────────────────────────────────────────────────────
@@ -806,26 +1021,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 +1119,14 @@ 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)
+    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 +1143,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
@@ -1086,7 +1283,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..a83b729 100644
--- a/packages/meshbay-node/src/meshbay_node/roots.py
+++ b/packages/meshbay-node/src/meshbay_node/roots.py
@@ -120,10 +120,10 @@ class Root:
     name: str
     path: Path
     kind: str = "generic"
-    upload: bool = False
+    writable: bool = False
+    removable: 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.
+    ejected: bool = False
     available: bool = True
 
     @property
@@ -172,8 +172,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 +210,18 @@ 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)))
             root = Root(name=name, path=path, kind=kind,
-                        upload=bool(spec.get("upload", False)),
+                        writable=writable,
+                        removable=bool(spec.get("removable", False)),
                         direct=bool(spec.get("direct", 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 +232,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 +318,23 @@ 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 and not root.ejected:
+                root.ejected = True
+                log.warning("Root %r auto-ejected (device disappeared): %s",
+                            root.name, root.path)
             if live != root.available:
                 root.available = live
                 changed.append((root, live))
@@ -352,7 +347,12 @@ class RootSet:
         out = []
         for r in self.roots:
             d: dict = {"name": r.name, "kind": r.kind,
-                        "available": r.available, "upload": r.upload}
+                        "available": r.available,
+                        "writable": r.writable,
+                        "removable": r.removable,
+                        "ejected": r.ejected,
+                        # Backward compat for MNP 1.0 clients
+                        "upload": r.writable}
             if r.direct:
                 d["direct"] = True
             out.append(d)
diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py
index c78281b..e2f749f 100644
--- a/packages/meshbay-node/src/meshbay_node/roster.py
+++ b/packages/meshbay-node/src/meshbay_node/roster.py
@@ -587,11 +587,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]:
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..4d6dd34 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -78,6 +78,9 @@ from meshbay_common.adminop import (
     OP_PHOTO_ROOTS,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
+    OP_ROOT_UPDATE,
+    OP_ROOT_EJECT,
+    OP_ROOT_PLUG,
     OP_GROUP_ATTACH,
     OP_GROUP_DETACH,
     admin_transcript,
@@ -507,6 +510,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 +763,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
@@ -1756,52 +1767,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
@@ -1828,6 +1799,8 @@ class WebRTCPeerSession:
             self._send({"type": "error",
                        "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"})
             return
+        if "files" not in apps:
+            apps.append("files")
         if not self._has_admin_authority():
             self._send({"type": "error", "detail": "No authorized key for this"})
             return
@@ -2409,7 +2382,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 +2399,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 +2448,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.
@@ -3678,35 +3788,44 @@ 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 target root. If absent, pick the first writable
+        # one (backward compat with old clients that don't send it).
+        target_root_name = msg.get("root")
+        upload_root = None
+        if target_root_name:
+            from meshbay_common.paths import fold
+            target_folded = fold(target_root_name)
+            for r in roots:
+                if fold(r.name) == target_folded:
+                    upload_root = r
+                    break
+        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 found for uploads",
+                        "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",
                         "filename": filename})
             return
@@ -3981,8 +4100,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 +4140,15 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_ROOT_REMOVE:
             self._spawn(
                 self._admin_exec_root_remove(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/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index 6fdc78f..3d24000 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -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,13 +419,13 @@ def create_ui_app(state: dict) -> FastAPI:
             "current_dir": progress.current_dir,
         }
 
-    # ── Upload toggle (operator only, localhost) ─────────────────────────
+    # ── Upload toggle (DEPRECATED — per-root writable replaces this) ────
 
     @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)),
-        ))
+        log.warning("PUT member-upload is deprecated — use PATCH roots/{name} "
+                    "with writable instead")
+        return {"deprecated": True, "message": "Use per-root writable flag"}
 
     # ── Enabled apps (operator only, localhost) ────────────────────────────
     #