feat(node): several named roots per group, and one implementation per operation

Stage A — a group's content is a set of named roots
---------------------------------------------------
`shared_dir` becomes a list of {name, path, kind}. The name is the directory's
basename, derived once at add time and *stored*: recomputing it would
re-identify a whole library the day someone renames a folder on disk. Duplicate
names are refused case-insensitively and no root may contain another — both
compared with NFC folding, because most of these directories live on exFAT or
NTFS where `Films` and `films` are one directory.

Every index path carries its root name, in a one-root group as much as in a
five-root one. One path shape has to be got right once; two have to be kept
right for ever.

**A root that goes away freezes; it never empties.** Unmounting a volume makes
watchdog report every file under it as deleted, or presents an empty directory
to the next scan. Acting on either propagates deletions for a whole library to
every member, as though the owner had erased it. So a deletion is acted on only
once its root is confirmed readable, and availability is tracked per root — one
unplugged drive leaves the others serving. 12 tests, verified to fail against an
indexer without the check.

Events are not trusted to be complete either: ReadDirectoryChangesW drops them
under load and inotify on a FUSE mount misses changes made outside it. A
periodic reconciliation sweep is the only thing that recovers a missed event.

MNP 0.2 → 0.3 (additive). The hub needs no change: SwarmSource carries a content
hash, a node id and an endpoint — no paths, no filenames — and private groups
register nothing (H7).

Stage B — one implementation behind every front door
----------------------------------------------------
C1 and C6 were both "a second path into the node with its own weaker
handshake". Two implementations of `revoke` with two authorization checks is
that shape one size down. `meshbay_node/ops.py` holds each operation once,
takes the daemon state, and knows nothing about HTTP, argv or MNP. The loopback
API is one `_op(...)` line per endpoint; the MNP handlers call the same
functions. test_ops.py asserts the shape rather than trusting it.

Phase 14 is finished on top of it — `group list`, `gek init|rotate`, `reload`
(SIGHUP), `denylist show|clear`, `file list|rm`. **No operator action requires a
browser any more.** Plus `gek_rotate` and `member_unpin` as operator-signed MNP
operations: rotation is the half of revocation that revocation cannot do, since
the ex-member holds the current key, and the node generates the replacement
with its own CSPRNG — no key material crosses the wire, which is what the C5b
rule is actually about.

Two bugs found by running it rather than by testing it
------------------------------------------------------
GroupIndex is keyed by **content hash**, so the same bytes at two paths are one
entry — which is also why a scan reports ten files and indexes nine.
Reconciliation compared paths, so it decided the second path was a missed event
every 60 s, rewrote the entry and pushed an index update to every connected
peer. Seen in a live node's log.

`meshbay-node reload` crashed on first use with `subprocess` unimported: the
module compiles fine, which is the "syntax, not names" trap already recorded for
the SPA. test_cli_dispatch.py now walks every verb and refuses to let one be
added to the parser without an entry there.

Also corrected: protocol.py declared a second MNP_VERSION of "0.1" while the
wire carried "0.2" — harmless only because nothing imported it. And
_do_dir_create/_do_dir_delete referenced an undefined `filename` on their error
path.

740 tests pass; QE/deploy/e2e.py passes end to end against the live deployment.

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

2 files changed, 270 insertions, 71 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
index 73e668c..ce6fe17 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -35,6 +35,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived, StreamReset
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common import MNP_VERSION
+from meshbay_node.roots import RootSet, entry_abs_path
 from meshbay_common.handshake import (
     NONCE_LEN,
     ROLE_CLIENT,
@@ -98,6 +99,37 @@ class Denylist:
         log.info("Denied jti: %s", jti[:8])
         self._save()
 
+    def entries(self) -> dict[str, list[str]]:
+        """What is currently refused, for the operator to inspect (14.10)."""
+        return {
+            "users": sorted(self.user_ids),
+            "groups": sorted(self.group_ids),
+            "jtis": sorted(self.jtis),
+        }
+
+    def clear(self, subject: str = "") -> int:
+        """
+        Drop everything, or one identifier. Returns how many entries went.
+
+        Not silent by design: clearing re-admits whoever it was keeping out, and
+        the count is what tells the operator whether they undid one revocation
+        or all of them.
+        """
+        before = len(self.user_ids) + len(self.group_ids) + len(self.jtis)
+        if subject:
+            self.user_ids.discard(subject)
+            self.group_ids.discard(subject)
+            self.jtis.discard(subject)
+        else:
+            self.user_ids.clear()
+            self.group_ids.clear()
+            self.jtis.clear()
+        after = len(self.user_ids) + len(self.group_ids) + len(self.jtis)
+        removed = before - after
+        if removed:
+            self._save()
+        return removed
+
     def _load(self) -> None:
         if not self._path or not self._path.exists():
             return
@@ -352,7 +384,7 @@ class _MNPServerProtocol(QuicConnectionProtocol):
             self._send(stream_id, {"type": "error", "detail": "File not found"})
             return
 
-        file_path = ctx["shared_root"] / entry.path / entry.name
+        file_path = entry_abs_path(ctx["roots"], entry)
         if not file_path.exists():
             self._send(stream_id, {"type": "error", "detail": "File not on disk"})
             return
@@ -379,7 +411,7 @@ class _MNPServerProtocol(QuicConnectionProtocol):
             self._send(stream_id, {"type": "error", "detail": "File not found"})
             return
 
-        file_path = ctx["shared_root"] / entry.path / entry.name
+        file_path = entry_abs_path(ctx["roots"], entry)
         if not file_path.exists():
             self._send(stream_id, {"type": "error", "detail": "File not on disk"})
             return
@@ -506,7 +538,7 @@ class QuicChunkServer:
         sk_node: Ed25519PrivateKey,
         hub_pk_pem: bytes,
         gek: bytes,
-        shared_root: Path,
+        roots: RootSet,
         index: GroupIndex,
         host: str = "::",   # listen IPv4 + IPv6 (dual-stack Linux)
         port: int = 19000,
@@ -519,7 +551,7 @@ class QuicChunkServer:
             "sk_node":     sk_node,
             "hub_pk_pem":  hub_pk_pem,
             "gek":         gek,
-            "shared_root": shared_root,
+            "roots": roots,
             "index":       index,
         }
         if groups:
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 6b87e31..4ec841f 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -60,6 +60,8 @@ from meshbay_common.adminop import (
     OP_FILE_DELETE,
     OP_INVITE_CREATE,
     OP_MEMBER_REVOKE,
+    OP_GEK_ROTATE,
+    OP_MEMBER_UNPIN,
     admin_transcript,
 )
 from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
@@ -72,6 +74,8 @@ from meshbay_common.join import (
 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import GroupIndex
+from meshbay_node import ops
+from meshbay_node.roots import RootSet, entry_abs_path
 from meshbay_node.roster import DEFAULT_INVITE_TTL
 
 log = logging.getLogger(__name__)
@@ -156,37 +160,47 @@ def _free_name(directory: Path, filename: str) -> str:
     raise FileExistsError(filename)
 
 
-def safe_subdir(shared_root: Path, rel: str) -> Path | None:
+def safe_subdir(roots: RootSet, rel: str) -> Path | None:
     """
-    Resolve a client-supplied directory under the shared root, or refuse.
+    Resolve a client-supplied directory inside one of the group's roots, or refuse.
 
-    Uploads land where the member is looking now rather than in a per-user
-    quarantine, so the path arrives from the wire and every part of it has to be
-    checked: each segment against the same allowlist as filenames, and the
-    resolved result against the root. `..`, absolute paths, symlinks pointing
-    out, and anything with a separator in a segment are all refused here rather
-    than in the caller, so there is one place to get it right.
+    The path arrives from the wire, so every part is checked: the first segment
+    must name a root that is readable right now, each later segment against the
+    same allowlist as filenames, and the resolved result against that root's
+    directory. `..`, absolute paths, symlinks pointing out, and anything with a
+    separator in a segment are all refused here rather than in the caller, so
+    there is one place to get it right.
+
+    The virtual root itself — `""` — is deliberately **not** resolvable. It is
+    not a directory on anyone's disk: a file cannot be written there and a
+    directory cannot be created there, because it belongs to no volume. Callers
+    that used to receive the shared root for an empty path now receive None,
+    which is the honest answer.
 
     The quarantine was the fix for C5a; what actually mattered in it — no
     overwrite, a name allowlist, and confinement — is kept by this plus the
     caller's existing checks.
     """
-    rel = (rel or "").strip().strip("/")
-    if not rel:
-        return shared_root
-    parts = [seg for seg in rel.split("/") if seg not in ("", ".")]
+    found = roots.split(rel or "")
+    if found is None:
+        return None
+    root, tail = found
+    if not root.available:
+        return None
+    parts = [seg for seg in tail.split("/") if seg not in ("", ".")]
     if any(seg == ".." or not SAFE_UPLOAD_NAME.match(seg) for seg in parts):
         return None
     try:
-        target = (shared_root / Path(*parts)).resolve()
-        root = shared_root.resolve()
+        target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve()
+        base = root.path.resolve()
     except OSError:
         return None
-    if target != root and root not in target.parents:
+    if target != base and base not in target.parents:
         return None
     return target
 
 
+
 def _extract_dtls_fingerprint(sdp: str) -> bytes:
     """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
     for line in sdp.splitlines():
@@ -439,6 +453,10 @@ class WebRTCPeerSession:
                 self._do_invite_create(msg)
             elif mtype == MNP.MEMBER_REVOKE:
                 self._do_member_revoke(msg)
+            elif mtype == MNP.MEMBER_UNPIN:
+                self._do_member_unpin(msg)
+            elif mtype == MNP.GEK_ROTATE:
+                self._do_gek_rotate(msg)
             elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
                 self._spawn(self._do_keypair_bundle_store(msg))
             elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
@@ -1023,10 +1041,9 @@ class WebRTCPeerSession:
         but it writes to the operator's disk, so it is audited like one.
         """
         ctx = self._group_ctx()
-        shared_root = ctx.get("shared_root")
-        if not shared_root:
-            self._send({"type": "error", "detail": "No shared directory",
-                        "filename": filename})
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
+            self._send({"type": "error", "detail": "No shared directory"})
             return
 
         name = str(msg.get("name", "")).strip()
@@ -1034,12 +1051,21 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "Invalid directory name"})
             return
 
-        parent = safe_subdir(shared_root, msg.get("dir") or "")
+        # The virtual root is not a directory on anyone's disk, so a member
+        # cannot create one there — that would be adding a root, which is the
+        # operator's configuration and not a file operation.
+        parent_rel = (msg.get("dir") or "").strip("/")
+        if not parent_rel:
+            self._send({"type": "error",
+                        "detail": "Choose a folder to create this in"})
+            return
+
+        parent = safe_subdir(roots, parent_rel)
         if parent is None or not parent.is_dir():
             self._send({"type": "error", "detail": "Invalid directory"})
             return
 
-        target = safe_subdir(shared_root, f"{(msg.get('dir') or '').strip('/')}/{name}")
+        target = safe_subdir(roots, f"{parent_rel}/{name}")
         if target is None:
             self._send({"type": "error", "detail": "Invalid directory"})
             return
@@ -1048,14 +1074,20 @@ class WebRTCPeerSession:
             return
 
         target.mkdir(parents=False)
-        log.info("Directory created by %s: %s", self._user_id[:8],
-                 target.relative_to(shared_root))
-        self._audit("dir_create", str(target.relative_to(shared_root)))
+        virtual = roots.virtual_of(target) or f"{parent_rel}/{name}"
+        log.info("Directory created by %s: %s", self._user_id[:8], virtual)
+        self._audit("dir_create", virtual)
         self._send({
             "type": MNP.DIR_CREATE_ACK, "v": MNP_VERSION,
-            "dir": str(target.relative_to(shared_root)),
+            "dir": virtual,
         })
 
+    @staticmethod
+    def _names_a_root(roots: RootSet, rel: str) -> bool:
+        """True when `rel` is a bare root name rather than something inside one."""
+        found = roots.split(rel or "")
+        return found is not None and not found[1]
+
     def _do_dir_delete(self, msg: dict) -> None:
         """
         Remove an empty directory, for the node operator.
@@ -1068,14 +1100,17 @@ class WebRTCPeerSession:
         operator deletes the files first and sees what they are losing.
         """
         ctx = self._group_ctx()
-        shared_root = ctx.get("shared_root")
-        if not shared_root:
-            self._send({"type": "error", "detail": "No shared directory",
-                        "filename": filename})
+        roots: RootSet | None = ctx.get("roots")
+        if not roots:
+            self._send({"type": "error", "detail": "No shared directory"})
             return
 
-        target = safe_subdir(shared_root, msg.get("dir") or "")
-        if target is None or target == shared_root:
+        rel = (msg.get("dir") or "").strip("/")
+        target = safe_subdir(roots, rel)
+        # A root itself is not deletable here: removing one is a configuration
+        # change, and doing it through a file operation would leave the group
+        # config naming a directory nobody can reach.
+        if target is None or self._names_a_root(roots, rel):
             self._send({"type": "error", "detail": "Invalid directory"})
             return
         if not target.is_dir():
@@ -1089,16 +1124,17 @@ class WebRTCPeerSession:
             return
 
         self._issue_admin_challenge(
-            OP_DIR_DELETE, str(target.relative_to(shared_root)))
+            OP_DIR_DELETE, roots.virtual_of(target) or rel)
 
     async def _admin_exec_dir_delete(
         self, pending: dict, transcript: bytes, sig: bytes,
     ) -> None:
         rel = pending["subject"]
         ctx = self._group_ctx()
-        shared_root = ctx.get("shared_root")
-        target = safe_subdir(shared_root, rel) if shared_root else None
-        if target is None or target == shared_root or not target.is_dir():
+        roots: RootSet | None = ctx.get("roots")
+        target = safe_subdir(roots, rel) if roots else None
+        if (target is None or self._names_a_root(roots, rel)
+                or not target.is_dir()):
             self._send({"type": "error", "detail": "Not a directory"})
             return
 
@@ -1145,6 +1181,95 @@ class WebRTCPeerSession:
             return
         self._issue_admin_challenge(OP_MEMBER_REVOKE, user_id)
 
+    def _do_gek_rotate(self, msg: dict) -> None:
+        """
+        Ask for a new group key. Operator only, and signed.
+
+        This is what actually removes a revoked member's access: revocation
+        stops the node serving the *next* key, and they still hold the current
+        one. The node generates the replacement itself — nothing arriving here
+        contributes key material, which is what the C5b rule is about.
+        """
+        if not self._group_id:
+            self._send({"type": "error", "detail": "No group on this connection"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_GEK_ROTATE, self._group_id)
+
+    async def _admin_exec_gek_rotate(
+        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"gek_rotate:{pending['subject'][:8]}")
+            return
+        try:
+            result = await self._run_op(
+                ops.set_gek, pending["subject"], rotate=True)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("gek_rotate", pending["subject"])
+        self._send({
+            "type": MNP.GEK_ROTATE_ACK, "v": MNP_VERSION,
+            "group_id": pending["subject"],
+            "authorized_members": result.get("authorized_members", 0),
+            # Said plainly, because rotating is the step people skip: content
+            # already downloaded stays readable to whoever holds it.
+            "note": "members re-receive the key on their next connect; content "
+                    "already downloaded is unaffected",
+        })
+
+    def _do_member_unpin(self, msg: dict) -> None:
+        """Forget a pinned identity, so someone can pair again with a new key."""
+        user_id = str(msg.get("user_id", "")).strip()
+        if not user_id:
+            self._send({"type": "error", "detail": "Missing user_id"})
+            return
+        if user_id == self._user_id:
+            # Unpinning yourself over the connection your pin authorizes would
+            # end that connection's authority mid-operation.
+            self._send({"type": "error", "detail": "Cannot unpin yourself"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(OP_MEMBER_UNPIN, user_id)
+
+    async def _admin_exec_member_unpin(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        user_id = 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"member_unpin:{user_id[:8]}")
+            return
+        try:
+            await self._run_op(ops.unpin_member, user_id)
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        self._audit("member_unpin", user_id)
+        self._send({"type": MNP.MEMBER_UNPIN_ACK, "v": MNP_VERSION,
+                    "user_id": user_id})
+
+    async def _run_op(self, fn, *args, **kwargs):
+        """
+        Call an operation from `meshbay_node.ops` with the daemon's own view.
+
+        The transport carries its own context and the loopback API carries the
+        daemon state; they overlap but are not the same dict. Handing the MNP
+        path a *second* set of lookups is exactly how two implementations of one
+        operation start disagreeing — C1 and C6 one size down — so the daemon
+        publishes its state here and both adapters call the same function.
+        """
+        state = self._ctx.get("daemon_state")
+        if state is None:
+            raise ops.OpError("Node state not available", status=503)
+        return await fn(state, *args, **kwargs)
+
     async def _admin_exec_member_revoke(
         self, pending: dict, transcript: bytes, sig: bytes,
     ) -> None:
@@ -1285,24 +1410,39 @@ class WebRTCPeerSession:
             # Directories are not index entries, so the client used to infer them
             # from file paths — which means a folder someone just created, or one
             # they emptied, simply did not exist as far as the UI was concerned.
-            "dirs": self._list_dirs(ctx.get("shared_root")),
+            "dirs": self._list_dirs(ctx.get("roots")),
+            # Which top-level folders are roots, and whether each is readable.
+            # A frozen root's files stay listed, so without this a member cannot
+            # tell "the drive is unplugged" from "it is all still there".
+            "roots": ctx["roots"].describe() if ctx.get("roots") else [],
         })
 
     @staticmethod
-    def _list_dirs(shared_root: Path | None) -> list[str]:
-        """Directories under the shared root, relative and sorted."""
-        if not shared_root:
-            return []
-        out = []
-        try:
-            for path in sorted(shared_root.rglob("*")):
-                if path.is_dir() and not path.name.startswith("."):
-                    rel = path.relative_to(shared_root)
-                    if not any(part.startswith(".") for part in rel.parts):
-                        out.append(str(rel))
-        except OSError:
+    def _list_dirs(roots: RootSet | None) -> list[str]:
+        """
+        Every directory in the group, as members address them, sorted.
+
+        Each root appears as a directory in its own right, so a root holding no
+        files yet is still somewhere a member can navigate to and upload into.
+        An unavailable root is listed too — its content is frozen, not gone, and
+        hiding it would look exactly like deletion.
+        """
+        if not roots:
             return []
-        return out[:2000]
+        out: list[str] = []
+        for root in roots:
+            out.append(root.name)
+            if not root.available:
+                continue
+            try:
+                for path in sorted(root.path.rglob("*")):
+                    if path.is_dir() and not path.name.startswith("."):
+                        rel = path.relative_to(root.path)
+                        if not any(part.startswith(".") for part in rel.parts):
+                            out.append(f"{root.name}/{rel.as_posix()}")
+            except OSError:
+                continue
+        return sorted(out)[:2000]
 
     async def _do_file_request(self, msg: dict) -> None:
         ctx = self._group_ctx()
@@ -1314,7 +1454,7 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
 
-        file_path = ctx["shared_root"] / entry.path / entry.name
+        file_path = entry_abs_path(ctx["roots"], entry)
         if not file_path.exists():
             self._send({"type": "error", "detail": "File not on disk"})
             return
@@ -1372,7 +1512,7 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
 
-        file_path = ctx["shared_root"] / entry.path / entry.name
+        file_path = entry_abs_path(ctx["roots"], entry)
         if not file_path.exists():
             self._send({"type": "error", "detail": "File not on disk"})
             return
@@ -1542,20 +1682,41 @@ class WebRTCPeerSession:
                         "filename": filename})
             return
 
-        shared_root = ctx.get("shared_root")
-        if not shared_root:
-            self._send({"type": "error", "detail": "No shared directory",
+        roots: RootSet | None = ctx.get("roots")
+        upload_root = roots.upload_root if roots 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",
+                        "filename": filename})
+            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 "
+                                  f"currently unavailable",
                         "filename": filename})
             return
 
-        # One destination, chosen here and not by the client: uploads/ at the root
-        # of the shared directory. C5a is still honoured — the name passed the
-        # allowlist above, and an existing file is never replaced, which was the
-        # real defect (overwriting a file also made the attacker its recorded
-        # uploader, and therefore able to delete it).
-        rel_dir = UPLOAD_DIR_NAME
-        target_dir = shared_root / UPLOAD_DIR_NAME
-        target_dir.mkdir(parents=True, exist_ok=True)
+        # One destination, chosen by the operator and not by the client:
+        # uploads/ inside the group's designated root. C5a is still honoured —
+        # the name passed the allowlist above, and an existing file is never
+        # replaced, which was the real defect (overwriting a file also made the
+        # attacker its recorded uploader, and therefore able to delete it).
+        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",
+                        "filename": filename})
+            return
 
         upload_key = f"{rel_dir}/{filename}"
         state = self._uploads.get(upload_key)
@@ -1789,6 +1950,12 @@ class WebRTCPeerSession:
         elif pending["op"] == OP_INVITE_CREATE:
             self._spawn(
                 self._admin_exec_invite_create(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_GEK_ROTATE:
+            self._spawn(
+                self._admin_exec_gek_rotate(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_MEMBER_UNPIN:
+            self._spawn(
+                self._admin_exec_member_unpin(pending, transcript, sig_bytes))
         else:
             self._send({"type": "error", "detail": "Unknown admin operation"})
 
@@ -1865,7 +2032,7 @@ class WebRTCPeerSession:
         })
 
     def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
-        file_path = ctx["shared_root"] / entry.path / entry.name
+        file_path = entry_abs_path(ctx["roots"], entry)
         if file_path.exists():
             file_path.unlink()
             log.info("File deleted: %s", entry.name)
@@ -2049,7 +2216,7 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not found"})
             return
 
-        file_path = ctx["shared_root"] / entry.path / entry.name
+        file_path = entry_abs_path(ctx["roots"], entry)
         if not file_path.exists():
             self._send({"type": "error", "detail": "File not on disk"})
             return
@@ -2265,7 +2432,7 @@ class WebRTCTransport:
     Manages WebRTC peer connections for browser clients.
 
     Usage:
-        transport = WebRTCTransport(sk_node, hub_pk_pem, gek, shared_root, index)
+        transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, index)
         answer_sdp = await transport.handle_offer(offer_sdp, peer_id)
         # Return answer_sdp to the browser via hub signaling
     """
@@ -2275,7 +2442,7 @@ class WebRTCTransport:
         sk_node: Ed25519PrivateKey,
         hub_pk_pem: bytes,
         gek: bytes,
-        shared_root: Path,
+        roots: RootSet,
         index: GroupIndex,
         groups: dict[str, dict] | None = None,
         denylist: Any | None = None,
@@ -2286,7 +2453,7 @@ class WebRTCTransport:
             "sk_node": sk_node,
             "hub_pk_pem": hub_pk_pem,
             "gek": gek,
-            "shared_root": shared_root,
+            "roots": roots,
             "index": index,
             "_peers": {},
             # None means "the operator said nothing" — the default applies. It