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

2 files changed, 526 insertions, 93 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..8e357c9 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -76,8 +76,14 @@ from meshbay_common.adminop import (
     OP_MUSICBRAINZ_ENABLED,
     OP_AUDIO_ROOT,
     OP_PHOTO_ROOTS,
+    OP_APP_DIRECTORIES,
+    OP_CHAT_DIRECTORY,
+    OP_CHAT_LINK_PREVIEW,
     OP_ROOT_ADD,
     OP_ROOT_REMOVE,
+    OP_ROOT_UPDATE,
+    OP_ROOT_EJECT,
+    OP_ROOT_PLUG,
     OP_GROUP_ATTACH,
     OP_GROUP_DETACH,
     admin_transcript,
@@ -208,7 +214,6 @@ JOIN_FAILURE_WINDOW = 600  # seconds
 # attachments from the chat alike. One visible directory the operator can look
 # into, back up or empty — rather than a hidden tree of per-user uuids that
 # nobody could read, or files scattered wherever someone happened to be looking.
-UPLOAD_DIR_NAME = "uploads"
 
 
 def _extract_dtls_fingerprint(sdp: str) -> bytes:
@@ -481,6 +486,12 @@ class WebRTCPeerSession:
                 self._do_audio_root(msg)
             elif mtype == MNP.PHOTO_ROOTS:
                 self._do_photo_roots(msg)
+            elif mtype == MNP.APP_DIRECTORIES:
+                self._do_app_directories(msg)
+            elif mtype == MNP.CHAT_DIRECTORY:
+                self._do_chat_directory(msg)
+            elif mtype == MNP.CHAT_LINK_PREVIEW:
+                self._do_chat_link_preview(msg)
             elif mtype == MNP.MEDIA_META_REQ:
                 self._spawn(self._do_media_meta_request(msg))
             elif mtype == MNP.SEASON_META_REQ:
@@ -507,6 +518,12 @@ class WebRTCPeerSession:
                 self._do_root_add(msg)
             elif mtype == MNP.ROOT_REMOVE:
                 self._do_root_remove(msg)
+            elif mtype == MNP.ROOT_UPDATE:
+                self._do_root_update(msg)
+            elif mtype == MNP.ROOT_EJECT:
+                self._do_root_eject(msg)
+            elif mtype == MNP.ROOT_PLUG:
+                self._do_root_plug(msg)
             elif mtype == MNP.ROSTER_READ:
                 self._spawn(self._do_roster_read(msg))
             elif mtype == MNP.DENYLIST_READ:
@@ -754,10 +771,12 @@ class WebRTCPeerSession:
         # channel and nothing else.
         config = {
             "is_node_admin": self._is_node_admin(),
-            # So the interface knows whether to offer uploading at all. Not a
-            # permission — the node refuses regardless — but without it the
-            # only way to discover the answer is to try.
-            "member_upload": bool(self._group_ctx().get("member_upload", True)),
+            # Backward compat for MNP 1.0 clients: computed from writable roots.
+            # New clients read per-root writable from the index payload instead.
+            "member_upload": any(
+                r.get("writable") for r in
+                (self._group_ctx().get("roots").describe()
+                 if self._group_ctx().get("roots") else [])),
             # Which group "applications" to show. Absent/empty falls back to
             # every registered one client-side, so a node that predates this
             # setting (or one whose context has not loaded it yet) hides
@@ -791,6 +810,23 @@ class WebRTCPeerSession:
             # (docs/photos.md §2.1). Empty means unset (the Photos tab shows
             # nothing yet).
             "photo_roots": list(self._group_ctx().get("photo_roots") or []),
+            # The same three answers in one shape, plus every other app's —
+            # `<app>_directories`, keyed by the app's registry name, always a
+            # list. The scalars above are derived from these (daemon.py's
+            # `_app_directories_ctx`) and kept for MNP 1.0 clients, which can
+            # represent one folder and never more. A client reading the plural
+            # form gets all of them.
+            **{f"{app}_directories":
+               list(self._group_ctx().get(f"{app}_directories") or [])
+               for app in ("video", "music", "photo", "chat")},
+            # Where chat attachments are written — the singular form, because
+            # Chat genuinely has one destination. "" means the operator has not
+            # chosen, and the paperclip says so.
+            "chat_directory": self._group_ctx().get("chat_directory") or "",
+            # Whether the node unfurls links members post here. Absent means
+            # on, which is what it did before this existed.
+            "chat_link_preview": bool(
+                self._group_ctx().get("chat_link_preview", True)),
             # So a client that connects mid-scan shows the indexing state
             # immediately, instead of waiting for the next periodic
             # INDEX_PROGRESS push. Never a path or filename — see
@@ -1559,6 +1595,28 @@ class WebRTCPeerSession:
                         "detail": "Choose a folder to create this in"})
             return
 
+        # Read-only means read-only, and creating a folder writes to the
+        # operator's disk. `_do_file_upload` gained this check with the RO/RW
+        # model and this one did not — so a member could not add a file to a
+        # published library but could still leave empty directories in it.
+        owner = roots.split(parent_rel)
+        if owner is None:
+            self._send({"type": "error", "detail": "Invalid directory"})
+            return
+        parent_root, _tail = owner
+        if not parent_root.writable:
+            self._send({"type": "error",
+                        "detail": f"Directory '{parent_root.name}' is read-only",
+                        "code": "root_read_only"})
+            self._audit("dir_create_refused", parent_rel[:64])
+            return
+        if not parent_root.available:
+            self._send({"type": "error",
+                        "detail": f"Directory '{parent_root.name}' is "
+                                  f"currently unavailable",
+                        "code": "root_unavailable"})
+            return
+
         parent = safe_subdir(roots, parent_rel)
         if parent is None or not parent.is_dir():
             self._send({"type": "error", "detail": "Invalid directory"})
@@ -1756,52 +1814,12 @@ class WebRTCPeerSession:
                     "user_id": user_id})
 
     def _do_member_upload(self, msg: dict) -> None:
-        """
-        Turn uploading by ordinary members on or off, for this group.
-
-        Signed like every other operator action. The setting decides who may
-        write to the operator's disk, so a node that took it from an unsigned
-        message would let any member turn it back on for everyone — the control
-        would be a suggestion.
-        """
-        if "allowed" not in msg:
-            self._send({"type": "error", "detail": "Missing allowed"})
-            return
-        if not self._has_admin_authority():
-            self._send({"type": "error", "detail": "No authorized key for this"})
-            return
-        # The subject is what the operator is shown before signing, so it has to
-        # name the outcome rather than the operation.
-        self._issue_admin_challenge(
-            OP_MEMBER_UPLOAD, "on" if msg.get("allowed") else "off")
-
-    async def _admin_exec_member_upload(
-        self, pending: dict, transcript: bytes, sig: bytes,
-    ) -> None:
-        allowed = pending["subject"] == "on"
-        if not await self._verify_admin_sig(transcript, sig):
-            self._send({"type": "error", "detail": "Signature verification failed"})
-            self._audit("admin_auth_failed", f"member_upload:{pending['subject']}")
-            return
-        try:
-            await self._run_op(
-                ops.set_member_upload, self._group_id or "", allowed)
-        except ops.OpError as e:
-            self._send({"type": "error", "detail": e.message})
-            return
-        self._audit("member_upload", pending["subject"])
-
-        # Everyone already connected is told, rather than finding out by having
-        # an upload refused. Enforcement does not depend on this reaching them —
-        # it is the node that refuses — but a button that stays visible until
-        # the next reconnection is a button people press.
-        notice = {"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
-                  "allowed": allowed}
-        for uid, session in list(self._peer_registry().items()):
-            try:
-                session._send(notice)
-            except Exception:
-                pass
+        # Deprecated: upload control is now per-root via writable flag.
+        # Old clients may still send this — acknowledge without acting.
+        log.warning("Deprecated member_upload message received — use root "
+                    "writable/read-only instead")
+        self._send({"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
+                    "allowed": True, "deprecated": True})
 
     # Every "application" a group can show. Photos joins this set (and
     # apps.js's registry, client-side) when it lands; nothing else about
@@ -1810,13 +1828,20 @@ class WebRTCPeerSession:
     # network calls (TMDB, MusicBrainz) once enabled, so an operator opts a
     # group in explicitly rather than getting it for free
     # (docs/mediacenter.md §5.6, docs/musicbay.md §4.4).
-    ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo"})
+    # `helloworld` is the reference implementation (docs/refactor-groups.md
+    # §4.1), hidden client-side behind `?dev=1`. It is here because the
+    # allow-list is server-side enforcement — a client that names an app this
+    # node does not know is refused — and an app the node refused could not
+    # demonstrate anything. This entry and the client's registry line are the
+    # whole of what adding an application costs.
+    ALLOWED_APPS = frozenset({"chat", "files", "video", "music", "photo",
+                              "helloworld"})
 
     def _do_apps_enabled(self, msg: dict) -> None:
         """
         Turn a group "application" on or off for everyone, for this group.
 
-        Signed like `member_upload`: this decides what a member sees, and an
+        Signed like the root ops: this decides what a member sees, and an
         unsigned message would let any member turn a disabled one back on.
         """
         apps = msg.get("apps")
@@ -1828,6 +1853,12 @@ class WebRTCPeerSession:
             self._send({"type": "error",
                        "detail": f"Unknown app(s): {', '.join(sorted(unknown))}"})
             return
+        # Files is not a toggle: MNP permits root exploration regardless of
+        # what this list says, so hiding the tab only ever misled. Added at the
+        # front, the same order ops.set_enabled_apps writes, so the landing-tab
+        # preference sees one list and not two.
+        if "files" not in apps:
+            apps.insert(0, "files")
         if not self._has_admin_authority():
             self._send({"type": "error", "detail": "No authorized key for this"})
             return
@@ -1911,7 +1942,7 @@ class WebRTCPeerSession:
         self._audit("tmdb_config", pending["subject"])
 
         # Node-wide setting: every connected peer in every group is told, not
-        # just this group's peers (unlike apps_enabled/member_upload/the
+        # just this group's peers (unlike apps_enabled/the root ops/the
         # per-group tmdb_enabled below).
         notice = {
             "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION,
@@ -2119,6 +2150,150 @@ class WebRTCPeerSession:
             except Exception:
                 pass
 
+    # ── App directories (generic) ────────────────────────────────────────
+
+    def _do_app_directories(self, msg: dict) -> None:
+        """
+        Which folder(s) an application works over, for any application.
+
+        One handler where there were three near-identical ones (`video_root`,
+        `audio_root`, `photo_roots`) differing only in the key they wrote and
+        whether they carried a string or a list. Those three still exist for
+        clients that speak them; nothing new is added beside them.
+
+        `app` must be one this node knows (`ALLOWED_APPS`) — a client-supplied
+        key is otherwise a way to write arbitrary rows into `group_settings`.
+        The paths are checked by `ops._validate_app_dirs`, which runs after the
+        signature: this is a settings change, not a capability, so refusing
+        early here would be a courtesy rather than the control.
+        """
+        app = str(msg.get("app", "")).strip()
+        dirs = msg.get("directories")
+        if app not in self.ALLOWED_APPS:
+            self._send({"type": "error", "detail": f"Unknown app {app!r}"})
+            return
+        if not isinstance(dirs, list) or not all(isinstance(d, str) for d in dirs):
+            self._send({"type": "error",
+                        "detail": "Missing or invalid 'directories'"})
+            return
+        clean = sorted({d.strip("/") for d in dirs if d.strip("/")})
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        # The app is in the subject, not only the paths: an operator shown
+        # "Media/Films" alone cannot tell which application is about to be
+        # pointed at it, and two apps' challenges would be indistinguishable.
+        self._issue_admin_challenge(
+            OP_APP_DIRECTORIES, f"{app}:{','.join(clean)}")
+
+    async def _admin_exec_app_directories(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        app, _, joined = pending["subject"].partition(":")
+        dirs = joined.split(",") if joined else []
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"app_directories:{pending['subject']}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_app_directories, self._group_id or "", app, dirs)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("app_directories", pending["subject"])
+        self._broadcast_to_group({"type": MNP.APP_DIRECTORIES_ACK,
+                                  "v": MNP_VERSION, "app": app,
+                                  "directories": result["directories"]})
+
+    # ── Chat ─────────────────────────────────────────────────────────────
+
+    def _do_chat_directory(self, msg: dict) -> None:
+        """
+        Where chat attachments are written.
+
+        Unlike every other app directory this one is a destination, so it has
+        to be on a read-write root — checked by `ops.set_chat_directory` after
+        the signature, which is where the refusal actually lives.
+        """
+        path = msg.get("path")
+        if not isinstance(path, str):
+            self._send({"type": "error", "detail": "Missing or invalid 'path'"})
+            return
+        path = path.strip("/")
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_CHAT_DIRECTORY, path)
+
+    async def _admin_exec_chat_directory(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        path = pending["subject"]
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"chat_directory:{path}")
+            return
+        try:
+            await self._run_op(
+                ops.set_chat_directory, self._group_id or "", path)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_directory", path)
+        self._broadcast_to_group(
+            {"type": MNP.CHAT_DIRECTORY_ACK, "v": MNP_VERSION, "path": path})
+
+    def _do_chat_link_preview(self, msg: dict) -> None:
+        """
+        Whether the node fetches a page's title and image when a member posts
+        a link — outbound traffic on the operator's connection, from a message
+        they did not write, so it is signed like everything else that decides
+        what leaves this machine.
+        """
+        enabled = msg.get("enabled")
+        if not isinstance(enabled, bool):
+            self._send({"type": "error", "detail": "Missing or invalid 'enabled'"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_CHAT_LINK_PREVIEW, "on" if enabled else "off")
+
+    async def _admin_exec_chat_link_preview(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        enabled = pending["subject"] == "on"
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"chat_link_preview:{pending['subject']}")
+            return
+        try:
+            await self._run_op(
+                ops.set_chat_link_preview, self._group_id or "", enabled)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("chat_link_preview", pending["subject"])
+        self._broadcast_to_group({"type": MNP.CHAT_LINK_PREVIEW_ACK,
+                                  "v": MNP_VERSION, "enabled": enabled})
+
+    def _broadcast_to_group(self, notice: dict) -> None:
+        """
+        Tell everyone connected to this group about a setting that changed.
+
+        Enforcement never depends on this reaching them — the node is what
+        refuses — but a control that stays on screen until the next
+        reconnection is a control people use.
+        """
+        for _uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     def _do_musicbrainz_enabled(self, msg: dict) -> None:
         """
         Whether MusicBrainz lookups run for this group at all. Per-group
@@ -2308,11 +2483,14 @@ class WebRTCPeerSession:
         if not self._has_admin_authority():
             self._send({"type": "error", "detail": "No authorized key for this"})
             return
-        upload_dir = str(msg.get("upload_dir", "")).strip()
+        # `upload_dir` is not read here any more, and a client still sending it
+        # is ignored rather than obeyed: on load it forces every other root
+        # read-only, which is the model the RO/RW one replaced. A second
+        # writable directory is `root_add` with `writable`.
         self._issue_admin_challenge(
             OP_GROUP_ATTACH, name,
             payload={"name": name, "shared_dir": shared_dir,
-                     "upload_dir": upload_dir},
+                     "writable": bool(msg.get("writable", True))},
             group_id="")
 
     async def _admin_exec_group_attach(
@@ -2326,7 +2504,8 @@ class WebRTCPeerSession:
         p = pending.get("payload") or {}
         try:
             result = await self._run_op(
-                ops.attach_group, p["name"], p["shared_dir"], p.get("upload_dir", ""))
+                ops.attach_group, p["name"], p["shared_dir"],
+                writable=bool(p.get("writable", True)))
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
@@ -2409,7 +2588,8 @@ class WebRTCPeerSession:
                 "group_id": target_group, "path": path,
                 "name": str(msg.get("name", ""))[:128],
                 "kind": str(msg.get("kind", "generic"))[:16],
-                "upload": bool(msg.get("upload", False)),
+                "writable": bool(msg.get("writable", msg.get("upload", False))),
+                "removable": bool(msg.get("removable", False)),
             },
             group_id=target_group)
 
@@ -2425,7 +2605,8 @@ class WebRTCPeerSession:
             result = await self._run_op(
                 ops.add_root, p["group_id"], p["path"],
                 name=p.get("name", ""), kind=p.get("kind", "generic"),
-                upload=p.get("upload", False))
+                writable=p.get("writable", False),
+                removable=p.get("removable", False))
         except ops.OpError as e:
             self._send({"type": "error", "detail": e.message})
             return
@@ -2473,6 +2654,141 @@ class WebRTCPeerSession:
         await self._retarget_indexer(p["group_id"])
         self._send({"type": MNP.ROOT_REMOVE_ACK, "v": MNP_VERSION, **result})
 
+    def _do_root_update(self, msg: dict) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        updates = []
+        if "writable" in msg:
+            updates.append(f"rw={'on' if msg['writable'] else 'off'}")
+        if "removable" in msg:
+            updates.append(f"rem={'on' if msg['removable'] else 'off'}")
+        subject = f"{root_name}:{','.join(updates)}" if updates else root_name
+        self._issue_admin_challenge(
+            OP_ROOT_UPDATE, subject,
+            payload={
+                "group_id": target_group, "root_name": root_name,
+                "writable": msg.get("writable"),
+                "removable": msg.get("removable"),
+            },
+            group_id=target_group)
+
+    async def _admin_exec_root_update(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"root_update:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(
+                ops.update_root, p["group_id"], p["root_name"],
+                writable=p.get("writable"), removable=p.get("removable"))
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("root_update failed: %s", e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit("root_update", pending["subject"])
+        await self._retarget_indexer(p["group_id"])
+        notice = {"type": MNP.ROOT_UPDATE_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    def _do_root_eject(self, msg: dict) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_ROOT_EJECT, root_name,
+            payload={"group_id": target_group, "root_name": root_name},
+            group_id=target_group)
+
+    async def _admin_exec_root_eject(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"root_eject:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(
+                ops.eject_root, p["group_id"], p["root_name"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("root_eject failed: %s", e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit("root_eject", p["root_name"])
+        notice = {"type": MNP.ROOT_EJECT_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
+    def _do_root_plug(self, msg: dict) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            OP_ROOT_PLUG, root_name,
+            payload={"group_id": target_group, "root_name": root_name},
+            group_id=target_group)
+
+    async def _admin_exec_root_plug(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed",
+                        f"root_plug:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(
+                ops.plug_root, p["group_id"], p["root_name"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("root_plug failed: %s", e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit("root_plug", p["root_name"])
+        notice = {"type": MNP.ROOT_PLUG_ACK, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     async def _run_op(self, fn, *args, **kwargs):
         """
         Call an operation from `meshbay_node.ops` with the daemon's own view.
@@ -2489,10 +2805,36 @@ class WebRTCPeerSession:
         return await fn(state, *args, **kwargs)
 
     async def _retarget_indexer(self, group_id: str) -> None:
-        """Tell the indexer to rescan after roots changed."""
+        """
+        Pick up a root that was just added to or removed from node.toml.
+
+        Through the daemon's own reload, which is what the loopback API has
+        always done after the same operations (`ui/app.py`). This used to
+        re-point the indexer at `groups_ctx[gid]["roots"]` instead — the very
+        object the op had just edited — so `retarget` diffed a set against
+        itself, found no new names, scanned nothing, and dropped nothing. A
+        directory added over MNP reached node.toml and was invisible until a
+        restart; one removed kept serving its files.
+
+        Two front doors doing different things is the shape `ops.py` exists to
+        prevent, and this was it: the loopback path worked and the MNP path did
+        not, which is why it survived until the operator added a directory from
+        a browser.
+
+        Not awaited: a reload rescans, and a new library is minutes. The ack
+        the caller sends carries the set the node is moving to, and the
+        `index_sync` that follows the scan carries what it found.
+        """
         state = self._ctx.get("daemon_state")
         if not state:
             return
+        reload_fn = state.get("reload_fn")
+        if reload_fn:
+            self._spawn(reload_fn())
+            return
+        # No daemon to ask — a test harness, or a context assembled by hand.
+        # Retarget directly, which is correct as long as the caller did not
+        # edit the live set in place.
         indexer = state.get("indexers", {}).get(group_id)
         roots = state.get("groups_ctx", {}).get(group_id, {}).get("roots")
         if indexer and roots:
@@ -3620,6 +3962,17 @@ class WebRTCPeerSession:
         """
         url = msg.get("url")
         key = url if isinstance(url, str) else ""
+
+        # Checked before the cache, not after: the operator turning previews
+        # off has to stop serving the ones already fetched too, or the setting
+        # takes effect only for links nobody has posted yet. Refused as an
+        # ordinary miss — the client shows the bare link, which is exactly what
+        # "no preview" looks like for a page that has none.
+        if not self._group_ctx().get("chat_link_preview", True):
+            self._send({"type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION,
+                        "url": key, "ok": False})
+            return
+
         cached = _link_preview_cache_get(key)
         if cached is not None:
             self._send({**cached, "type": MNP.LINK_PREVIEW_RESP, "v": MNP_VERSION})
@@ -3678,51 +4031,101 @@ class WebRTCPeerSession:
                         "filename": filename})
             return
 
-        # The operator can close uploading to everyone but themselves. Enforced
-        # here rather than by hiding a button: the button is a courtesy to the
-        # people who are not trying, and this is the part that holds against
-        # someone who is. `is_node_admin` is computed from the identity this
-        # node pinned, never from a hub claim.
-        if not ctx.get("member_upload", True) and not self._is_node_admin():
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
             self._send({"type": "error",
-                        "detail": "Uploading is turned off for this group",
-                        "code": "member_upload_off",
+                        "detail": "No directories configured for this group",
                         "filename": filename})
-            self._audit("upload_refused", filename[:64])
             return
 
-        roots: RootSet | None = ctx.get("roots")
-        upload_root = roots.upload_root if roots else None
+        # The client names the root it is uploading into — it is browsing one,
+        # and with several writable roots any other choice is a guess. It names
+        # a root, never a path: the destination inside it is decided below and
+        # is not negotiable, which is what keeps C5a closed.
+        #
+        # An unknown name is refused rather than falling back to a writable
+        # root, because "the file went somewhere else" is discovered weeks
+        # later — the same reason the old single upload root was never guessed.
+        # A client that names nothing is an MNP 1.0 one, and there was exactly
+        # one destination in its world: the first writable root.
+        # `dir` is the folder being browsed, as a virtual path
+        # (`Media/Films/1999`); `root` is the older, coarser form and is what
+        # its first segment means on its own.
+        target_rel = str(msg.get("dir") or "").strip().strip("/")
+        target_root_name = (target_rel.split("/")[0] if target_rel
+                            else str(msg.get("root") or "").strip())
+        upload_root = None
+        if target_root_name:
+            upload_root = roots.by_name(target_root_name)
+            if upload_root is None:
+                self._send({"type": "error",
+                            "detail": f"No directory named "
+                                      f"{target_root_name!r} in this group",
+                            "code": "no_such_root",
+                            "filename": filename})
+                return
+        else:
+            writable = roots.writable_roots
+            upload_root = writable[0] if writable else None
+
         if upload_root is None:
-            # Refused, never guessed. With several roots, picking one would send
-            # a member's file to a disk the operator did not intend, and that is
-            # discovered weeks later.
             self._send({"type": "error",
-                        "detail": "No upload folder is configured for this group",
+                        "detail": "No writable directory in this group",
+                        "code": "no_writable_root",
+                        "filename": filename})
+            return
+        if not upload_root.writable:
+            self._send({"type": "error",
+                        "detail": f"Directory '{upload_root.name}' is read-only",
+                        "code": "root_read_only",
                         "filename": filename})
+            self._audit("upload_refused", filename[:64])
             return
         if not upload_root.available:
-            # The designated root's volume is absent. Falling back to another
-            # root would scatter uploads across disks depending on what happened
-            # to be plugged in.
             self._send({"type": "error",
-                        "detail": f"The upload folder ({upload_root.name}) is "
+                        "detail": f"Directory '{upload_root.name}' is "
                                   f"currently unavailable",
+                        "code": "root_unavailable",
                         "filename": filename})
             return
 
-        if upload_root.direct:
-            rel_dir = upload_root.name
-            target_dir = upload_root.path
+        # The folder the sender is looking at, and no subdirectory of the node's
+        # invention.
+        #
+        # Uploads used to be confined to `<root>/uploads/`, created on demand.
+        # That was the last of v5's quarantine (the per-user layer went on
+        # 2026-08-14, for the same reason): a shared directory nobody can
+        # organise is not a shared directory, and a folder appearing beside the
+        # operator's library because somebody sent a file is the node deciding
+        # how their disk is arranged.
+        #
+        # What made the quarantine worth having is not the subdirectory — it is
+        # the filename allowlist, the size cap, the chunk ordering, and the
+        # no-overwrite rule below. All four are unchanged.
+        #
+        # `resolve()` and not a join: it refuses `..`, absolute segments and
+        # anything whose resolved form escapes its root, symlinks included. The
+        # client names *where among the group's own folders*, never a path on
+        # the operator's filesystem.
+        if target_rel:
+            target_dir = roots.resolve(target_rel)
+            if target_dir is None or not target_dir.is_dir():
+                self._send({"type": "error",
+                            "detail": "Not a directory in this group",
+                            "code": "no_such_directory",
+                            "filename": filename})
+                return
+            rel_dir = target_rel
         else:
-            rel_dir = f"{upload_root.name}/{UPLOAD_DIR_NAME}"
-            target_dir = upload_root.path / UPLOAD_DIR_NAME
-            try:
-                target_dir.mkdir(parents=True, exist_ok=True)
-            except OSError as e:
-                log.warning("Cannot create upload folder in root %r: %s",
-                            upload_root.name, e)
-                self._send({"type": "error", "detail": "Upload folder unavailable",
+            # An MNP 1.0 client names nothing; the root itself is where its one
+            # destination now is.
+            target_dir = upload_root.path
+            rel_dir = upload_root.name
+            if not target_dir.is_dir():
+                self._send({"type": "error",
+                            "detail": f"Directory '{upload_root.name}' is "
+                                      f"currently unavailable",
+                            "code": "root_unavailable",
                             "filename": filename})
                 return
 
@@ -3981,8 +4384,10 @@ class WebRTCPeerSession:
             self._spawn(
                 self._admin_exec_member_unpin(pending, transcript, sig_bytes))
         elif pending["op"] == OP_MEMBER_UPLOAD:
-            self._spawn(
-                self._admin_exec_member_upload(pending, transcript, sig_bytes))
+            log.warning("Deprecated OP_MEMBER_UPLOAD signed op — use root "
+                        "writable/read-only instead")
+            self._send({"type": MNP.MEMBER_UPLOAD_ACK, "v": MNP_VERSION,
+                        "allowed": True, "deprecated": True})
         elif pending["op"] == OP_APPS_ENABLED:
             self._spawn(
                 self._admin_exec_apps_enabled(pending, transcript, sig_bytes))
@@ -4019,6 +4424,24 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_ROOT_REMOVE:
             self._spawn(
                 self._admin_exec_root_remove(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_APP_DIRECTORIES:
+            self._spawn(
+                self._admin_exec_app_directories(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_CHAT_DIRECTORY:
+            self._spawn(
+                self._admin_exec_chat_directory(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_CHAT_LINK_PREVIEW:
+            self._spawn(
+                self._admin_exec_chat_link_preview(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_ROOT_UPDATE:
+            self._spawn(
+                self._admin_exec_root_update(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_ROOT_EJECT:
+            self._spawn(
+                self._admin_exec_root_eject(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_ROOT_PLUG:
+            self._spawn(
+                self._admin_exec_root_plug(pending, transcript, sig_bytes))
         elif pending["op"] == OP_GROUP_ATTACH:
             self._spawn(
                 self._admin_exec_group_attach(pending, transcript, sig_bytes))
diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py
index c683204..6986b01 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/wire.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py
@@ -89,13 +89,21 @@ def index_sync_message(index, roots: RootSet | None) -> dict:
     }
 
 
-def index_delta_message(index, delta) -> dict:
+def index_delta_message(index, delta, roots=None) -> dict:
     """
     One `index_delta` — what changed since the last thing this node broadcast.
 
     Built here rather than inline in the daemon, which is where it lived and which
     made it the third place an index message was constructed: precisely the drift
     that produced two `index_sync` encodings and two `file_chunk` encodings before it.
+
+    `roots` rides along (MNP 1.1, additive — a 1.0 client ignores it). It used
+    to travel on `index_sync` alone, which is a *full* index and therefore only
+    ever sent on request. So a root added, removed, ejected or plugged left
+    every connected client's directory table stale until somebody reloaded the
+    page: the delta that told them something had changed was the one message
+    that could not say what. It is a handful of dicts, bounded by the number of
+    directories a group has, and it is sealed with the rest.
     """
     payload = {
         "base_version": delta.base_version,
@@ -104,6 +112,8 @@ def index_delta_message(index, delta) -> dict:
         "deletions": list(delta.deletions),
         "updates": [index_entry_wire(e) for e in delta.updates],
     }
+    if roots is not None:
+        payload["roots"] = roots.describe()
     return {
         "type": MNP.INDEX_DELTA,
         "v": MNP_VERSION,