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>

1 files changed, 203 insertions, 73 deletions

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