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>

9 files changed, 1863 insertions, 415 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index 3de2473..42ebcfd 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -47,13 +47,29 @@ max_concurrent_streams = 8
 # Browser and native clients reach this node over WebRTC DataChannel via hub
 # signaling — no inbound port to open. QUIC is the optional direct path.
 
-# Multiple groups — each with its own directory
+# Multiple groups — each with one or more named directories ("roots").
+#
+# 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.
 [[groups]]
 id          = ""    # set after joining
 name        = "My Media"
-shared_dir  = "/home/user/Media"
 quic_port   = 19010
 
+  [[groups.roots]]
+  path   = "/home/user/Media"
+  upload = 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
+
+# The single-directory form still works and means the same thing — one root,
+# named after the directory, receiving uploads.
 [[groups]]
 id          = ""
 name        = "Public Archive"
@@ -97,10 +113,24 @@ class NodeConfig:
 
 
 @dataclass
+class RootSpec:
+    """One named directory inside a group. See `meshbay_node/roots.py`."""
+    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
+
+
+@dataclass
 class GroupConfig:
     id:         str = ""
     name:       str = ""
-    shared_dir: str = ""
+    # A group's content is several named roots. `shared_dir` is the single-root
+    # form and is still read: it becomes one root named after its basename, which
+    # is why every path gained a segment. See roots.py for why there is no
+    # unprefixed shape.
+    roots:      list[RootSpec] = field(default_factory=list)
+    shared_dir: str = ""      # legacy single-root form, migrated at load
     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
@@ -112,6 +142,19 @@ class GroupConfig:
     join_policy: str = "invite"   # invite|open
     quic_port:  int = 19010       # QUIC MNP port
 
+    def __post_init__(self) -> None:
+        """
+        The single-directory form becomes one root, whoever built this.
+
+        On the dataclass rather than in the TOML reader, because a GroupConfig is
+        also built by the CLI, by `group attach` and by tests. Putting the
+        migration in the parser alone left every one of those paths with a group
+        that had no directory at all — and it presented as "skipping group",
+        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)]
+
 
 @dataclass
 class KeystoreConfig:
@@ -157,6 +200,35 @@ def _positive(value: object, default: int, name: str) -> int:
     return n
 
 
+def _read_roots(group: dict) -> list[RootSpec]:
+    """
+    A group's roots, from `[[groups.roots]]` or from the legacy `shared_dir`.
+
+    Both forms are accepted and `shared_dir` is not deprecated for a single
+    directory — it is the same thing said shorter. Naming both is refused rather
+    than merged: which one receives uploads would be a guess, and a wrong guess
+    is discovered weeks later.
+    """
+    specs = [
+        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 []
+    ]
+    legacy = str(group.get("shared_dir", "") or "").strip()
+    if specs and legacy:
+        log.warning(
+            "group %r declares both shared_dir and [[groups.roots]] — using "
+            "roots and ignoring shared_dir = %s",
+            group.get("name", ""), legacy)
+    # A bare shared_dir needs no handling here: GroupConfig.__post_init__ turns
+    # it into one root for every construction path, not just this one.
+    return specs
+
+
 def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
     """
     Load config from TOML file. Supports both single [group] and
@@ -190,7 +262,10 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
                 cfg.groups.append(GroupConfig(
                     id=g.get("id", ""),
                     name=g.get("name", ""),
-                    shared_dir=g.get("shared_dir", ""),
+                    roots=_read_roots(g),
+                    # Ignored when roots are given explicitly (warned about in
+                    # _read_roots); otherwise __post_init__ migrates it.
+                    shared_dir="" if _read_roots(g) else g.get("shared_dir", ""),
                     visibility=g.get("visibility", "private"),
                     join_policy=g.get("join_policy", "invite"),
                     quic_port=g.get("quic_port", cfg.node.quic_port),
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 8b4a1d7..21331f2 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -26,6 +26,7 @@ Usage:
 """
 
 import asyncio
+from dataclasses import asdict
 import base64
 import json
 import logging
@@ -42,6 +43,7 @@ from meshbay_node.audit import AuditStore
 from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
+from meshbay_node.roots import RootSet, RootError
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.keystore import load_or_create_keystore
@@ -197,17 +199,30 @@ class NodeDaemon:
             # 4. Build per-group contexts
             groups_ctx: dict[str, dict] = {}
             for group_cfg in self._config.groups:
-                if not group_cfg.id or not group_cfg.shared_dir:
-                    log.warning("Group %r missing id or shared_dir — skipping",
-                                group_cfg.name)
+                if not group_cfg.id or not group_cfg.roots:
+                    log.warning("Group %r has no id or no shared directory — "
+                                "skipping", group_cfg.name)
                     continue
 
-                shared_root = Path(group_cfg.shared_dir).expanduser().resolve()
-                if not shared_root.exists():
-                    log.warning("Shared dir not found: %s — skipping group %s",
-                                shared_root, group_cfg.name)
+                try:
+                    roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                except RootError as e:
+                    # Configuration the operator has to fix; guessing would put
+                    # a member's file on the wrong disk or index one twice.
+                    log.error("Group %r: %s — skipping", group_cfg.name, e)
                     continue
 
+                roots.refresh_availability()
+                if not any(r.available for r in roots):
+                    # Not skipped for being empty: a group whose only drive is
+                    # unplugged still exists, and its index is frozen rather
+                    # than lost. But there is nothing to serve until it returns.
+                    log.warning(
+                        "Group %r: none of its %d root(s) are readable right now "
+                        "(%s) — serving nothing until one returns",
+                        group_cfg.name, len(roots),
+                        ", ".join(str(r.path) for r in roots))
+
                 gek = None
                 if group_cfg.visibility == "private":
                     gek = await self._load_gek(
@@ -219,7 +234,7 @@ class NodeDaemon:
                                  group_cfg.name)
 
                 indexer = DirectoryIndexer(
-                    root=shared_root,
+                    roots=roots,
                     group_id=group_cfg.id,
                     sk_node=keys.sk_ed25519,
                     gek=gek,
@@ -229,11 +244,13 @@ class NodeDaemon:
                 self._indexers.append(indexer)
                 self._state["indexes"][group_cfg.id] = indexer.index
                 log.info("Indexing group %s: %s (%d files)",
-                         group_cfg.name, shared_root, indexer.index.count)
+                         group_cfg.name,
+                         ", ".join(f"{r.name}={r.path}" for r in roots),
+                         indexer.index.count)
 
                 groups_ctx[group_cfg.id] = {
                     "gek": gek,
-                    "shared_root": shared_root,
+                    "roots": roots,
                     "index": indexer.index,
                     "visibility": group_cfg.visibility,
                     # Admission policy comes from node.toml, never from the hub:
@@ -270,7 +287,7 @@ class NodeDaemon:
                     sk_node=keys.sk_ed25519,
                     hub_pk_pem=session.hub_pk_pem,
                     gek=first["gek"],
-                    shared_root=first["shared_root"],
+                    roots=first["roots"],
                     index=first["index"],
                     groups=groups_ctx,
                     denylist=denylist,
@@ -290,6 +307,11 @@ class NodeDaemon:
                 self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64
 
                 self._webrtc._ctx["roster"] = self._roster
+                # The MNP adapter calls the same operations as the loopback API
+                # (meshbay_node.ops), and those take the daemon's state. Handing
+                # the transport a second set of lookups is how two paths to one
+                # operation start disagreeing — the shape of C1 and C6.
+                self._webrtc._ctx["daemon_state"] = self._state
                 self._webrtc._ctx["invite_ttl"] = (
                     self._config.node.invite_ttl_hours * 3600)
                 paired = await self._roster.has_operator() if self._roster else False
@@ -310,7 +332,7 @@ class NodeDaemon:
                     sk_node=keys.sk_ed25519,
                     hub_pk_pem=session.hub_pk_pem,
                     gek=first["gek"],
-                    shared_root=first["shared_root"],
+                    roots=first["roots"],
                     index=first["index"],
                     host="::",
                     port=self._config.node.quic_port,
@@ -383,7 +405,13 @@ class NodeDaemon:
             self._state["roster"] = self._roster
             self._state["node_user_id"] = session.user_id
             self._state["webrtc"] = self._webrtc
+            self._state["quic_server"] = self._quic_server
             self._state["hub"] = hub
+            # Rotating a key has to reach every transport holding a copy of it,
+            # and clearing the denylist has to reach the one the handshake
+            # consults — so both are published rather than reachable only
+            # through the object that happens to own them.
+            self._state["denylist"] = self._denylist
             self._state["pk_x25519_raw"] = pk_x_raw
 
             self._state["status"] = "running"
@@ -410,10 +438,78 @@ class NodeDaemon:
             loop = asyncio.get_event_loop()
             for sig in (signal.SIGINT, signal.SIGTERM):
                 loop.add_signal_handler(sig, stop_event.set)
+            # Milestone 14.8: re-read node.toml without dropping connections.
+            try:
+                loop.add_signal_handler(
+                    signal.SIGHUP,
+                    lambda: asyncio.ensure_future(self._reload_config()))
+            except (NotImplementedError, AttributeError):
+                pass          # no SIGHUP on Windows; `reload` says so there
             await stop_event.wait()
 
         await self._shutdown()
 
+    async def _reload_config(self) -> None:
+        """
+        Re-read node.toml on SIGHUP.
+
+        Deliberately narrow: it picks up **root changes on groups already
+        hosted**, which is what an operator adjusts day to day, and reports
+        anything else as needing a restart. Adding or removing a whole group
+        means new indexers, chat stores, GEK loads and transport contexts, and
+        doing that under a live daemon is how a half-built group ends up serving
+        content. Saying "restart for that" is honest and costs one restart.
+
+        Nothing here touches connections: a member watching a film keeps
+        watching it.
+        """
+        log.info("SIGHUP — re-reading %s", self._config_path)
+        try:
+            fresh = load_config(self._config_path)
+        except Exception as e:
+            log.error("Reload failed, keeping the running config: %s", e)
+            return
+
+        groups_ctx = self._state.get("groups_ctx") or {}
+        hosted = set(groups_ctx)
+        incoming = {g.id for g in fresh.groups if g.id}
+        if incoming != hosted:
+            added = ", ".join(sorted(incoming - hosted)) or "none"
+            removed = ", ".join(sorted(hosted - incoming)) or "none"
+            log.warning("Group set changed (added: %s, removed: %s) — restart the "
+                        "daemon for that; roots of existing groups reloaded anyway",
+                        added, removed)
+
+        changed = 0
+        for group_cfg in fresh.groups:
+            ctx = groups_ctx.get(group_cfg.id)
+            if not ctx:
+                continue
+            try:
+                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+            except RootError as e:
+                log.error("Group %r: %s — keeping the roots already loaded",
+                          group_cfg.name, e)
+                continue
+            before = {(r.name, str(r.path)) for r in ctx["roots"]}
+            after = {(r.name, str(r.path)) for r in roots}
+            if before == after:
+                continue
+            roots.refresh_availability()
+            indexer = next((i for i in self._indexers
+                            if i.group_id == group_cfg.id), None)
+            if indexer is None:
+                continue
+            log.info("Group %r roots changed: %s", group_cfg.name,
+                     ", ".join(f"{r.name}={r.path}" for r in roots))
+            await indexer.retarget(roots)
+            ctx["roots"] = roots
+            changed += 1
+
+        self._config = fresh
+        self._state["config"] = fresh
+        log.info("Reload complete — %d group(s) re-rooted", changed)
+
     async def _login_with_retry(self, hub: HubClient):
         """Login to hub, retrying if the node key hasn't been linked yet."""
         import httpx as _httpx
@@ -675,21 +771,27 @@ def main() -> None:
 
     parser = argparse.ArgumentParser(description="MeshBay Node daemon")
     parser.add_argument("command", nargs="?",
-                        choices=["init", "status", "ui", "gek-init", "operator",
-                                 "member", "group", "calibrate-argon2"],
+                        choices=["init", "status", "ui", "gek-init", "gek",
+                                 "operator", "member", "group", "file",
+                                 "denylist", "reload", "calibrate-argon2"],
                         help="init: write example config | status: node state and keys "
                              "| ui: print the admin UI URL | operator pair: pair a "
                              "browser with this node | member list|invite|revoke|unpin "
-                             "| group add <name> --dir <path>: host another of your "
-                             "groups | calibrate-argon2: benchmark")
+                             "| group list|add | gek init|rotate | file list|rm "
+                             "| denylist show|clear | reload: re-read node.toml "
+                             "| calibrate-argon2: benchmark")
     parser.add_argument("subcommand", nargs="?",
                         help="'pair' for operator; list|invite|revoke|unpin for "
-                             "member; 'add' for group")
+                             "member; list|add for group; init|rotate for gek; "
+                             "list|rm for file; show|clear for denylist")
     parser.add_argument("target", nargs="?",
-                        help="username for member invite|revoke|unpin; "
-                             "group name for group add")
+                        help="username for member invite|revoke|unpin; group name "
+                             "for group add; file id for file rm; identifier for "
+                             "denylist clear")
     parser.add_argument("--dir", default=None,
                         help="shared directory, for group add")
+    parser.add_argument("--yes", action="store_true",
+                        help="skip the confirmation for destructive commands")
     parser.add_argument("--config", type=Path, default=None,
                         help="Config file path")
     parser.add_argument("--group", default=None,
@@ -699,8 +801,8 @@ def main() -> None:
     args = parser.parse_args()
 
     # Query commands print a report; library logging would interleave with it.
-    quiet = args.command in ("status", "ui", "gek-init", "operator", "member",
-                             "group")
+    quiet = args.command in ("status", "ui", "gek-init", "gek", "operator",
+                             "member", "group", "file", "denylist", "reload")
     logging.basicConfig(
         level=logging.ERROR if quiet else getattr(logging, args.log_level),
         format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
@@ -755,11 +857,17 @@ def main() -> None:
         print(f"config       {DEFAULT_CONFIG_PATH}")
         if not cfg.groups:
             print("groups       none configured — create a group on the hub, then add")
-            print("             a [[groups]] entry with its id and shared_dir")
+            print("             a [[groups]] entry with its id and a directory")
         else:
             for g in cfg.groups:
                 print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
-                print(f"             {g.shared_dir or '<no shared_dir>'}")
+                if not g.roots:
+                    print("             <no directory configured>")
+                for r in g.roots:
+                    label = r.name or Path(r.path).name
+                    flag = "  (uploads)" if r.upload else ""
+                    live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
+                    print(f"             {label} → {r.path}{flag}{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.
@@ -874,22 +982,172 @@ def main() -> None:
         print("usage: meshbay-node member list|invite|revoke|unpin")
         sys.exit(1)
 
-    if args.command == "gek-init":
+    if args.command in ("gek-init", "gek"):
+        # `gek-init` is the original spelling and still works. `gek rotate` is
+        # the one that matters after a revocation: the ex-member holds the
+        # current key and nothing else takes it from them.
+        sub = "init" if args.command == "gek-init" else (args.subcommand or "init")
+        if sub not in ("init", "rotate"):
+            print("usage: meshbay-node gek init|rotate [--group NAME]")
+            sys.exit(1)
+
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
         group_id = _resolve_group(cfg, args.group)
-        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek",
+
+        if sub == "rotate" and not args.yes:
+            print("Rotating replaces this group's key.")
+            print("  · every member re-receives it automatically on their next connect")
+            print("  · anyone revoked keeps the OLD key and loses access to new content")
+            print("  · content already downloaded stays readable to whoever has it")
+            if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek"
+                               f"{'?rotate=true' if sub == 'rotate' else ''}",
                           method="POST", timeout=60)
 
-        print(f"GEK ready for {group_id}")
+        verb = "rotated" if out.get("rotated") else "ready"
+        print(f"GEK {verb} for {group_id}")
         print(f"  {out.get('authorized_members', 0)} authorized member(s) — each "
               f"receives the key on connect")
         for err in out.get("errors") or []:
             print(f"  ! {err}")
         return
 
+    if args.command == "reload":
+        # Milestone 14.8. The daemon re-reads node.toml; groups that appeared or
+        # whose roots changed are picked up without dropping live connections.
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        import os as _os
+        import signal as _signal
+        import subprocess as _subprocess
+        # `--` is pgrep's own end-of-options marker and must be its own argument;
+        # folded into the pattern it searches for a process literally called
+        # "-- -m …". Anchored to the end of the command line so it matches the
+        # daemon and never a shell that merely mentions it — the same trap
+        # deploy-node.sh documents, where an unanchored pattern kills the script.
+        pid_out = _subprocess.run(
+            ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"],
+            capture_output=True, text=True)
+        pids = [int(x) for x in pid_out.stdout.split()]
+        if not pids:
+            print("Node is not running — start it with: meshbay-node")
+            sys.exit(1)
+        for pid in pids:
+            _os.kill(pid, _signal.SIGHUP)
+        print(f"sent SIGHUP to {len(pids)} daemon process(es)")
+        print("watch the result:  tail -f /tmp/meshbay-node.log")
+        return
+
+    if args.command == "denylist":
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        sub = args.subcommand or "show"
+
+        if sub == "show":
+            out = _daemon_api(cfg, "/api/denylist")
+            total = out.get("count", 0)
+            if not total:
+                print("denylist     empty — nothing is being refused")
+                return
+            for kind in ("users", "groups", "jtis"):
+                for entry in out.get(kind, []):
+                    print(f"  {kind[:-1]:<6} {entry}")
+            print(f"\n{total} entr(y/ies). These survive a restart (finding H4).")
+            return
+
+        if sub == "clear":
+            if not args.yes:
+                what = args.target or "EVERY entry"
+                print(f"Clearing the denylist re-admits {what}.")
+                print("A revocation the hub sent will not come back on its own.")
+                if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"):
+                    print("cancelled")
+                    return
+            out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}",
+                              method="POST")
+            print(f"removed {out['removed']} entr(y/ies) ({out['subject']})")
+            return
+
+        print("usage: meshbay-node denylist show|clear [identifier] [--yes]")
+        sys.exit(1)
+
+    if args.command == "file":
+        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_id}/files")
+            files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"]))
+            if not files:
+                print("no files indexed")
+                return
+            for f in files:
+                print(f"  {f['id'][:12]}  {f['size']:>12}  {f['path']}/{f['name']}")
+            print(f"\n{len(files)} file(s). Remove one with: "
+                  f"meshbay-node file rm <id>")
+            return
+
+        if sub == "rm":
+            # Milestone 14.11 — the last operator action that needed a browser.
+            if not args.target:
+                print("usage: meshbay-node file rm <file-id> [--group NAME]")
+                sys.exit(1)
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/files",)
+            matches = [f for f in out.get("files", [])
+                       if f["id"].startswith(args.target)]
+            if not matches:
+                print(f"no file whose id starts with {args.target!r}")
+                sys.exit(1)
+            if len(matches) > 1:
+                print(f"{args.target!r} matches {len(matches)} files — be more specific:")
+                for f in matches[:10]:
+                    print(f"  {f['id'][:16]}  {f['path']}/{f['name']}")
+                sys.exit(1)
+            target = matches[0]
+            if not args.yes:
+                print(f"Delete {target['path']}/{target['name']} "
+                      f"({target['size']} bytes) from disk?")
+                print("This removes the file itself, not just the listing.")
+                if input("delete? [y/N] ").strip().lower() not in ("y", "yes"):
+                    print("cancelled")
+                    return
+            _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}",
+                        method="DELETE")
+            print(f"deleted {target['path']}/{target['name']}")
+            return
+
+        print("usage: meshbay-node file list|rm <id> [--group NAME] [--yes]")
+        sys.exit(1)
+
     if args.command == "group":
+        if args.subcommand in (None, "list"):
+            # Milestone 14.2.
+            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+            out = _daemon_api(cfg, "/api/groups")
+            groups = out.get("groups", [])
+            if not groups:
+                print("no groups hosted — add one with: "
+                      "meshbay-node group add <name> --dir <path>")
+                return
+            for g in groups:
+                key = "GEK" if g.get("has_gek") else "NO KEY"
+                print(f"  {g['name']}  [{g['visibility']}/{g.get('join_policy')}]  "
+                      f"{key}  {g['file_count']} file(s)  "
+                      f"{g.get('peers', 0)} peer(s)")
+                print(f"      {g['id']}")
+                for r in g.get("roots", []):
+                    flags = " (uploads)" if r.get("upload") else ""
+                    live = "" if r.get("available", True) else "  [UNAVAILABLE]"
+                    print(f"      root {r['name']}{flags}{live}")
+                if not g.get("has_gek"):
+                    print(f"      give it a key:  meshbay-node gek init "
+                          f"--group {g['name']}")
+            return
+
         if args.subcommand != "add":
-            print("usage: meshbay-node group add <name> --dir <path>")
+            print("usage: meshbay-node group list|add <name> --dir <path>")
             sys.exit(1)
         if not args.target or not args.dir:
             print("usage: meshbay-node group add <name> --dir <path>")
@@ -904,8 +1162,9 @@ def main() -> None:
         print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
         print(f"  shared_dir   {out['shared_dir']}")
         print()
-        print("Restart the daemon so it picks the group up, then give it a key:")
-        print(f"  meshbay-node gek-init --group {out['name']}")
+        print("Tell the daemon to re-read its config, then give the group a key:")
+        print("  meshbay-node reload")
+        print(f"  meshbay-node gek init --group {out['name']}")
         print()
         print("The key is this group's own — members of your other groups cannot")
         print("read it, and joining one says nothing about the other.")
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
index a69429c..5dbdc5d 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
@@ -59,6 +59,12 @@ class GroupIndex:
     sk_node:  Ed25519PrivateKey
     gek:      bytes | None = None   # None → public group (no encryption)
     version:  int = 1
+    # The group's roots and whether each is readable right now. Travels inside
+    # the encrypted payload because it names the operator's directories, and a
+    # member needs it to tell "temporarily unavailable" from "deleted" — a
+    # distinction the entries alone cannot carry, since an unavailable root's
+    # files are still listed. Absent in an index written before roots existed.
+    roots:    list = field(default_factory=list)
     _entries: dict = field(default_factory=dict, repr=False)  # id → IndexEntry
 
     # ── Entry management ──────────────────────────────────────────────────────
@@ -90,6 +96,7 @@ class GroupIndex:
         payload = msgpack.packb({
             "group_id": self.group_id,
             "version":  self.version,
+            "roots":    list(self.roots),
             "entries":  [asdict(e) for e in self.entries],
         }, use_bin_type=True)
 
@@ -170,6 +177,9 @@ class GroupIndex:
             sk_node=sk_node,
             gek=gek,
             version=payload["version"],
+            # Absent from an index written before roots existed; an empty list
+            # reads as "nothing known about availability", not "no roots".
+            roots=payload.get("roots") or [],
         )
         for e in payload["entries"]:
             idx.add_entry(IndexEntry(**e))
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index 60dc04b..d9b1d2c 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -1,15 +1,29 @@
 """
-Directory indexer — watches a directory and maintains a GroupIndex.
+Directory indexer — watches a group's roots and maintains a GroupIndex.
 
-Uses watchdog for filesystem events. On any change (create/modify/delete/move),
-the affected file is re-scanned and the GroupIndex is updated.
-File metadata (blake3 hash, size, type, duration) is computed on first scan.
-Heavy operations (hashing large files) run in a thread pool to avoid blocking.
+Uses watchdog for filesystem events. On any change the affected file is
+re-scanned and the GroupIndex is updated. File metadata (blake3 hash, size,
+type) is computed on first scan; hashing runs in a thread pool.
+
+Two properties are worth stating because they are what the code is shaped
+around, not incidental:
+
+**A root that goes away freezes; it never empties.** Unmounting a volume either
+makes watchdog emit a deletion for every file under it or presents an empty
+directory to the next scan. Both would propagate deletions for a whole library
+as though the owner had erased it, to every member. So a deletion is acted on
+only once the root it belongs to has been confirmed still readable, and a root
+that is not is marked unavailable with its entries left exactly where they are.
+
+**Events are not trusted to be complete.** `ReadDirectoryChangesW` drops events
+when its buffer overflows under a burst, and inotify on a FUSE mount misses
+changes made outside it. Most users are on Windows sharing from exFAT, so both
+apply. A periodic reconciliation scan is therefore not a belt-and-braces extra;
+it is the only thing that recovers a missed event.
 """
 
 import asyncio
 import logging
-import mimetypes
 import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
@@ -20,8 +34,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from watchdog.events import FileSystemEvent, FileSystemEventHandler
 from watchdog.observers import Observer
 
+from meshbay_common.paths import fold, find_fold_collisions, long_path
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import Root, RootSet
 
 log = logging.getLogger(__name__)
 
@@ -59,7 +75,19 @@ def _is_indexable(path: Path) -> bool:
 _HASH_CHUNK = 8 * 1024 * 1024   # 8 MB streaming hash chunks
 
 
-def _scan_file(root: Path, file_path: Path) -> IndexEntry | None:
+def _virtual_dir(root: Root, file_path: Path) -> str:
+    """
+    The directory a file appears in, as members see it: `"Films/2024"`.
+
+    The root name is the first segment for every root, including the only one of
+    a single-root group — one path shape has to be got right once, two have to
+    be kept right forever.
+    """
+    rel = file_path.parent.relative_to(root.path)
+    return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"
+
+
+def _scan_file(root: Root, file_path: Path) -> IndexEntry | None:
     """Compute IndexEntry for a file. Blocking — run in executor.
     Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.)
     don't require loading the whole file into memory."""
@@ -68,33 +96,33 @@ def _scan_file(root: Path, file_path: Path) -> IndexEntry | None:
     try:
         stat = file_path.stat()
         hasher = blake3.blake3()
-        with open(file_path, "rb") as f:
+        # long_path is a no-op off Windows; there it is what lets a deep media
+        # library past MAX_PATH.
+        with open(long_path(file_path), "rb") as f:
             while chunk := f.read(_HASH_CHUNK):
                 hasher.update(chunk)
-        file_id = hasher.hexdigest()
-        rel_path = str(file_path.parent.relative_to(root))
-        if rel_path == ".":
-            rel_path = ""
         return IndexEntry(
-            id=file_id,
+            id=hasher.hexdigest(),
+            # Stored exactly as the filesystem gave it: this is the string that
+            # opens the file. Normalization is for comparison only.
             name=file_path.name,
-            path=rel_path,
+            path=_virtual_dir(root, file_path),
             size=stat.st_size,
             type=_detect_type(file_path),
             added_at=int(stat.st_mtime),
         )
-    except (OSError, PermissionError) as e:
+    except (OSError, PermissionError, ValueError) as e:
         log.warning("Cannot index %s: %s", file_path, e)
         return None
 
 
 class DirectoryIndexer:
     """
-    Watches a directory and keeps a GroupIndex up to date.
+    Watches a group's roots and keeps a GroupIndex up to date.
 
     Usage:
         indexer = DirectoryIndexer(
-            root=Path("/home/user/shared"),
+            roots=RootSet.build([{"path": "/home/user/shared", "upload": True}]),
             group_id="my-group",
             sk_node=sk,
             gek=gek_bytes,
@@ -105,61 +133,147 @@ class DirectoryIndexer:
         await indexer.stop()
     """
 
+    # How often to re-check which roots are readable and reconcile the index
+    # against what is actually on disk. Not a poll for changes — a backstop for
+    # the events the OS did not deliver, and the way a re-plugged drive is
+    # noticed.
+    RECONCILE_SECS = 60.0
+
     def __init__(
         self,
-        root: Path,
+        roots: RootSet,
         group_id: str,
         sk_node: Ed25519PrivateKey,
         gek: bytes | None,
         on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None,
     ):
-        self.root     = root.resolve()
+        self.roots    = roots
         self.group_id = group_id
         self.sk_node  = sk_node
         self.gek      = gek
         self.on_change = on_change
 
         self._index    = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek)
+        self._index.roots = roots.describe()
         self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer")
         self._observer: Observer | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
+        self._reconciler: asyncio.Task | None = None
+        self._pending_timers: dict[str, asyncio.TimerHandle] = {}
 
     @property
     def index(self) -> GroupIndex:
         return self._index
 
+    # ── Root lookup ───────────────────────────────────────────────────────────
+
+    def _root_for(self, file_path: Path) -> Root | None:
+        """Which root a real path belongs to, longest match first."""
+        try:
+            resolved = file_path.resolve()
+        except OSError:
+            resolved = file_path
+        best: Root | None = None
+        for root in self.roots:
+            try:
+                resolved.relative_to(root.path)
+            except ValueError:
+                continue
+            if best is None or len(root.path.parts) > len(best.path.parts):
+                best = root
+        return best
+
     # ── Initial scan ──────────────────────────────────────────────────────────
 
     async def initial_scan(self) -> None:
-        """Scan the entire directory tree. Run once at startup."""
-        log.info("Scanning %s ...", self.root)
+        """Scan every available root. Run once at startup."""
+        self.roots.refresh_availability()
+        total = 0
+        for root in self.roots:
+            if not root.available:
+                log.warning("Root %r is not readable at startup (%s) — its files "
+                            "are not indexed yet and will appear when it returns",
+                            root.name, root.path)
+                continue
+            total += await self._scan_root(root)
+        self._index.version = int(time.time())
+        self._index.roots = self.roots.describe()
+        self._report_collisions()
+        log.info("Initial scan complete: %d files across %d root(s)",
+                 total, len(self.roots))
+
+    async def _scan_root(self, root: Root) -> int:
+        log.info("Scanning %s (root %r) ...", root.path, root.name)
         loop = asyncio.get_event_loop()
-        files = [p for p in self.root.rglob("*") if p.is_file()]
         count = 0
+        try:
+            files = [p for p in root.path.rglob("*") if p.is_file()]
+        except OSError as e:
+            log.warning("Cannot scan root %r: %s", root.name, e)
+            return 0
         for file_path in files:
             entry = await loop.run_in_executor(
-                self._executor, _scan_file, self.root, file_path)
+                self._executor, _scan_file, root, file_path)
             if entry:
                 self._index.add_entry(entry)
                 count += 1
-        self._index.version = int(time.time())
-        log.info("Initial scan complete: %d files indexed", count)
+        return count
+
+    def _report_collisions(self) -> None:
+        """
+        Names that are the same file on a case-insensitive filesystem.
+
+        Reported, never resolved: on ext4 both files exist and only the operator
+        knows which was meant. Left silent, the pair reaches somebody on Windows
+        who can save one of them.
+        """
+        by_dir: dict[str, list[str]] = {}
+        for entry in self._index.entries:
+            by_dir.setdefault(fold(entry.path), []).append(entry.name)
+        for folded_dir, names in by_dir.items():
+            for _, clashing in find_fold_collisions(names).items():
+                log.warning(
+                    "Names that differ only by case or accent form in %s: %s — "
+                    "these are one file on NTFS or exFAT, and a member on Windows "
+                    "can only keep one of them",
+                    folded_dir or "/", ", ".join(sorted(clashing)))
 
     # ── Watchdog integration ──────────────────────────────────────────────────
 
     async def start(self) -> None:
-        """Start initial scan + filesystem watcher."""
+        """Start initial scan + filesystem watcher + reconciler."""
         self._loop = asyncio.get_event_loop()
         await self.initial_scan()
+        self._start_observer()
+        self._reconciler = asyncio.create_task(self._reconcile_loop())
 
+    def _start_observer(self) -> None:
         handler = _WatchdogHandler(self)
         self._observer = Observer()
-        self._observer.schedule(handler, str(self.root), recursive=True)
+        watched = 0
+        for root in self.roots:
+            if not root.available:
+                continue
+            try:
+                self._observer.schedule(handler, str(root.path), recursive=True)
+                watched += 1
+            except OSError as e:
+                log.warning("Cannot watch root %r: %s", root.name, e)
         self._observer.start()
-        log.info("Watching %s for changes", self.root)
+        log.info("Watching %d root(s) for changes", watched)
 
     async def stop(self) -> None:
-        """Stop the filesystem watcher."""
+        """Stop the filesystem watcher and the reconciler."""
+        if self._reconciler:
+            self._reconciler.cancel()
+            try:
+                await self._reconciler
+            except asyncio.CancelledError:
+                pass
+            self._reconciler = None
+        for handle in self._pending_timers.values():
+            handle.cancel()
+        self._pending_timers.clear()
         if self._observer:
             self._observer.stop()
             self._observer.join()
@@ -167,6 +281,171 @@ class DirectoryIndexer:
         self._executor.shutdown(wait=False)
         log.info("Indexer stopped")
 
+    async def retarget(self, roots: RootSet) -> None:
+        """
+        Point this indexer at a new set of roots, without a restart (14.8).
+
+        Entries under a root that is gone from the config are dropped — the
+        operator removed it deliberately, which is not the same event as a
+        volume disappearing, and conflating the two is what §6.9 exists to
+        prevent. Roots that survive keep their entries; new ones are scanned.
+        """
+        old_names = {r.folded for r in self.roots}
+        new_names = {r.folded for r in roots}
+
+        for root in self.roots:
+            if root.folded not in new_names:
+                dropped = self._entries_under(root)
+                log.info("Root %r removed from the config — dropping %d entries",
+                         root.name, len(dropped))
+                for entry in dropped:
+                    self._index.remove_entry(entry.id)
+
+        self.roots = roots
+        roots.refresh_availability()
+        for root in roots:
+            if root.folded not in old_names and root.available:
+                await self._scan_root(root)
+
+        self._index.roots = roots.describe()
+        self._index.version = int(time.time())
+        self._restart_observer()
+        if self.on_change:
+            await self.on_change(self)
+
+    # ── Reconciliation ────────────────────────────────────────────────────────
+
+    async def _reconcile_loop(self) -> None:
+        while True:
+            try:
+                await asyncio.sleep(self.RECONCILE_SECS)
+                await self.reconcile()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                log.exception("Reconcile failed — continuing")
+
+    async def reconcile(self) -> None:
+        """
+        Re-check availability, and rescan roots that came back.
+
+        The only place a root's entries are dropped: when the root is readable
+        and the files are genuinely gone. A root that is not readable is left
+        untouched, which is the whole point.
+        """
+        changed = self.roots.refresh_availability()
+        touched = False
+
+        for root, available in changed:
+            if available:
+                log.info("Root %r is back — rescanning", root.name)
+                self._drop_root_entries(root)
+                await self._scan_root(root)
+                touched = True
+            else:
+                # Frozen: entries stay, marked unavailable to members through
+                # the roots table in the index payload.
+                log.warning("Root %r went away — %d entries frozen, not deleted",
+                            root.name, len(self._entries_under(root)))
+                touched = True
+
+        if changed:
+            self._restart_observer()
+
+        if await self._sweep_available_roots():
+            touched = True
+
+        if touched:
+            self._index.roots = self.roots.describe()
+            self._index.version = int(time.time())
+            if self.on_change:
+                await self.on_change(self)
+
+    async def _sweep_available_roots(self) -> bool:
+        """
+        Catch what the watcher missed: files gone, and files never announced.
+
+        Only touches roots that are readable right now — a root whose volume is
+        absent has nothing to compare against, and comparing anyway is exactly
+        the mistake this module exists to avoid.
+        """
+        loop = asyncio.get_event_loop()
+        changed = False
+        for root in self.roots:
+            if not root.available:
+                continue
+            try:
+                on_disk = {p.resolve() for p in root.path.rglob("*")
+                           if _is_indexable(p)}
+            except OSError as e:
+                log.warning("Cannot reconcile root %r: %s", root.name, e)
+                continue
+
+            known: dict[Path, str] = {}
+            for entry in self._entries_under(root):
+                abs_path = self._entry_path(root, entry)
+                if abs_path:
+                    known[abs_path] = entry.id
+
+            for missing in set(known) - on_disk:
+                # Duplicate content is handled without a special case here: the
+                # entry goes, and the add loop below re-indexes the surviving
+                # copy under its own path, because the id is then absent. A
+                # dedicated "find the survivor" lookup was written first and
+                # deleted — it rehashed every file under the root on any single
+                # deletion, and a test proved it changed nothing.
+                self._index.remove_entry(known[missing])
+                log.info("Reconcile: %s is gone", missing)
+                changed = True
+
+            for added in on_disk - set(known):
+                entry = await loop.run_in_executor(
+                    self._executor, _scan_file, root, added)
+                if not entry:
+                    continue
+                # The index is keyed by **content hash**, so two identical files
+                # at two paths are one entry and the path comparison above
+                # cannot see the second. Adding it anyway rewrites that entry's
+                # path every cycle, bumps the version, and pushes an index
+                # update to every connected peer once a minute — for ever.
+                # Measured on a live node: `clip.mp4` present at the root and in
+                # uploads/ with the same bytes.
+                if self._index.get_entry(entry.id) is not None:
+                    log.debug("Reconcile: %s duplicates content already indexed "
+                              "as %s — leaving the index alone",
+                              added, entry.id[:8])
+                    continue
+                self._index.add_entry(entry)
+                log.info("Reconcile: %s appeared (missed event)", added)
+                changed = True
+        return changed
+
+    def _entries_under(self, root: Root) -> list[IndexEntry]:
+        prefix = fold(root.name)
+        return [e for e in self._index.entries
+                if fold(e.path).split("/", 1)[0] == prefix]
+
+    def _drop_root_entries(self, root: Root) -> None:
+        for entry in self._entries_under(root):
+            self._index.remove_entry(entry.id)
+
+    @staticmethod
+    def _entry_path(root: Root, entry: IndexEntry) -> Path | None:
+        _, _, tail = entry.path.partition("/")
+        try:
+            return (root.path / tail / entry.name).resolve() if tail else \
+                   (root.path / entry.name).resolve()
+        except OSError:
+            return None
+
+    def _restart_observer(self) -> None:
+        """Re-schedule watches after roots appeared or disappeared."""
+        if self._observer:
+            self._observer.stop()
+            self._observer.join()
+            self._observer = None
+        self._start_observer()
+
     # ── Internal update ───────────────────────────────────────────────────────
 
     _DEBOUNCE_SECS = 2.0
@@ -175,39 +454,60 @@ class DirectoryIndexer:
         """Called from watchdog thread — schedule debounced async update."""
         if not self._loop:
             return
-        key = str(file_path.resolve())
-        self._loop.call_soon_threadsafe(
-            self._debounce, key, file_path, deleted)
+        key = str(file_path)
+        self._loop.call_soon_threadsafe(self._debounce, key, file_path, deleted)
 
     def _debounce(self, key: str, file_path: Path, deleted: bool) -> None:
-        if not hasattr(self, "_pending_timers"):
-            self._pending_timers: dict[str, asyncio.TimerHandle] = {}
         old = self._pending_timers.pop(key, None)
         if old:
             old.cancel()
-        handle = self._loop.call_later(
-            self._DEBOUNCE_SECS,
-            lambda: asyncio.ensure_future(self._update_entry(file_path, deleted)),
-        )
-        self._pending_timers[key] = handle
 
-    def _remove_by_path(self, file_path: Path) -> None:
-        """Remove any existing entries that match this file's path + name."""
-        resolved = file_path.resolve()
-        to_remove = [
-            e.id for e in self._index.entries
-            if (self.root / e.path / e.name).resolve() == resolved
-        ]
-        for fid in to_remove:
-            self._index.remove_entry(fid)
+        def fire() -> None:
+            self._pending_timers.pop(key, None)
+            asyncio.ensure_future(self._update_entry(file_path, deleted))
+
+        self._pending_timers[key] = self._loop.call_later(self._DEBOUNCE_SECS, fire)
+
+    def _remove_by_path(self, root: Root, file_path: Path) -> None:
+        """Remove any existing entries that point at this file."""
+        try:
+            resolved = file_path.resolve()
+        except OSError:
+            resolved = file_path
+        for entry in self._entries_under(root):
+            if self._entry_path(root, entry) == resolved:
+                self._index.remove_entry(entry.id)
 
     async def _update_entry(self, file_path: Path, deleted: bool) -> None:
-        self._remove_by_path(file_path)
+        root = self._root_for(file_path)
+        if root is None:
+            return
+
+        if deleted and not root.is_live():
+            # The volume went away rather than the file. Freeze: mark the root
+            # and touch nothing. Every other event for this root will arrive
+            # here too and be dropped the same way, which is the intent — one
+            # unplugged drive must not empty a library.
+            if root.available:
+                root.available = False
+                self._index.roots = self.roots.describe()
+                log.warning("Root %r disappeared — ignoring deletion events and "
+                            "freezing %d entries", root.name,
+                            len(self._entries_under(root)))
+                self._index.version = int(time.time())
+                if self.on_change:
+                    await self.on_change(self)
+            return
+
+        if not root.available:
+            return
+
+        self._remove_by_path(root, file_path)
 
         if not deleted:
             loop = asyncio.get_event_loop()
             entry = await loop.run_in_executor(
-                self._executor, _scan_file, self.root, file_path)
+                self._executor, _scan_file, root, file_path)
             if entry:
                 self._index.add_entry(entry)
                 log.debug("Indexed: %s (%s, %d bytes)",
diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py
new file mode 100644
index 0000000..c111581
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/ops.py
@@ -0,0 +1,487 @@
+"""
+Operator operations — one implementation, several front doors.
+
+Three things ask this node to act: the CLI (over the loopback admin API), the
+local admin UI, and — from Stage B3 — signed MNP messages from a paired client.
+They must agree, and the way to make them agree is not to write the operation
+three times and hope.
+
+**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
+the same shape one size down. So each operation lives here once, takes the
+daemon's `state`, and knows nothing about HTTP, argv or MNP. The adapters
+translate: `ui/app.py` turns `OpError` into a JSON response, the CLI prints it,
+the MNP handler sends an error frame.
+
+**Authorization is not here.** Reaching this module already means the caller got
+past its adapter's check — the loopback session token (11.5.3) for the API, an
+Ed25519 signature verified against the roster for MNP. These functions do what
+they are told; deciding who may tell them is the adapter's job and stays visible
+in the adapter.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any
+
+from meshbay_common.crypto import generate_gek, wrap_gek_aes
+from meshbay_node.config import DEFAULT_CONFIG_PATH
+from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
+from meshbay_node.roots import RootError, RootSet
+
+log = logging.getLogger(__name__)
+
+
+class OpError(Exception):
+    """
+    An operation refused, with enough for any adapter to report it.
+
+    `status` is an HTTP code because one adapter needs one; the others ignore it.
+    `extra` carries the "here is what would have worked" payload — a bare "no
+    such group" leaves an operator guessing at a UUID.
+    """
+
+    def __init__(self, message: str, *, status: int = 400,
+                 extra: dict[str, Any] | None = None):
+        super().__init__(message)
+        self.message = message
+        self.status = status
+        self.extra = extra or {}
+
+    def as_dict(self) -> dict:
+        return {"error": self.message, **self.extra}
+
+
+# ── Shared lookups ───────────────────────────────────────────────────────────
+
+def _roster(state: dict):
+    roster = state.get("roster")
+    if not roster:
+        raise OpError("Roster not available", status=503)
+    return roster
+
+
+def _hub(state: dict):
+    hub = state.get("hub")
+    if not hub or not hub._session:
+        raise OpError("Hub not connected", status=503)
+    return hub
+
+
+def _group_ctx(state: dict, group_id: str) -> dict:
+    groups_ctx = state.get("groups_ctx", {})
+    if group_id not in groups_ctx:
+        raise OpError("Group not hosted on this node", status=404,
+                      extra={"available": [
+                          {"id": gid} for gid in groups_ctx]})
+    return groups_ctx[group_id]
+
+
+def _config(state: dict):
+    config = state.get("config")
+    if not config:
+        raise OpError("No config loaded", status=503)
+    return config
+
+
+# ── Roster ───────────────────────────────────────────────────────────────────
+
+async def read_roster(state: dict, group_id: str = "") -> dict:
+    roster = state.get("roster")
+    if not roster:
+        return {"identities": [], "members": [], "invites": []}
+    return {
+        "identities": await roster.list_identities(),
+        "members": await roster.list_members(group_id or None),
+        "invites": await roster.list_invites(),
+    }
+
+
+async def resolve_user(state: dict, username: str) -> dict:
+    """
+    Map a username to an account id.
+
+    The roster answers first — it is the node's own record. The hub is the
+    fallback for identities pinned before invitations carried a name, and for
+    people admitted through an open-join group. Only an account id comes back;
+    no key is ever taken from there.
+    """
+    roster = state.get("roster")
+    if roster:
+        for ident in await roster.list_identities():
+            if ident["username"] == username:
+                return {"user_id": ident["user_id"], "source": "roster"}
+    hub = state.get("hub")
+    if hub and hub._session:
+        try:
+            account = await hub.get_user_pubkeys(username)
+            return {"user_id": account["user_id"], "source": "hub"}
+        except Exception:
+            pass
+    raise OpError(f"Unknown user {username!r}", status=404)
+
+
+async def pair_operator(state: dict) -> dict:
+    """
+    Issue a one-time code that pairs a browser as this node's operator.
+
+    The code is the whole point: it binds the operator's browser identity key to
+    their account without asking the hub, which is what stops a hub from naming
+    itself node administrator (M3, and the same substitution as H3). Returned
+    once and stored only as a hash.
+    """
+    roster = _roster(state)
+    user_id = state.get("node_user_id")
+    if not user_id:
+        raise OpError("Node not connected to hub yet", status=503)
+
+    config = state.get("config")
+    ttl = (config.node.pair_ttl_hours if config else 24) * 3600
+    code = await roster.create_invite(
+        group_id="",                 # operator authority is node-wide
+        user_id=user_id,
+        role=ROLE_OPERATOR,
+        created_by="local-cli",
+        ttl=ttl,
+        username=(config.hub.username if config else ""),
+    )
+    invites = await roster.list_invites()
+    expires = next((i["expires_at"] for i in invites
+                    if i["user_id"] == user_id and i["role"] == ROLE_OPERATOR), "")
+    return {"code": code, "expires_at": expires, "user_id": user_id}
+
+
+async def create_invite(state: dict, group_id: str, username: str) -> dict:
+    """
+    Issue an invitation code.
+
+    The hub is asked for the account id and nothing else — never for a key. A hub
+    that answered with the wrong account would produce an invite whose code it
+    never learns, since the code goes to a human out of band.
+    """
+    roster = _roster(state)
+    _group_ctx(state, group_id)
+    hub = _hub(state)
+    try:
+        account = await hub.get_user_pubkeys(username)
+    except Exception as e:
+        raise OpError(f"Unknown user {username!r}: {e}", status=404) from e
+
+    config = state.get("config")
+    ttl = (config.node.invite_ttl_hours if config else 168) * 3600
+    code = await roster.create_invite(
+        group_id=group_id,
+        user_id=account["user_id"],
+        role=ROLE_MEMBER,
+        created_by="local-cli",
+        ttl=ttl,
+        username=username,
+    )
+    invites = await roster.list_invites()
+    expires = next((i["expires_at"] for i in invites
+                    if i["user_id"] == account["user_id"]
+                    and i["group_id"] == group_id), "")
+    return {"code": code, "expires_at": expires,
+            "username": username, "user_id": account["user_id"]}
+
+
+async def revoke_member(state: dict, user_id: str, group_id: str) -> dict:
+    """
+    Stop serving the group key to someone.
+
+    Takes effect on their next connection: the key is wrapped on demand, so there
+    is no stored bundle left behind that would outlive this. Rotating the group
+    key is still required — they hold the current one.
+    """
+    roster = _roster(state)
+    if not await roster.set_status(group_id, user_id, "revoked"):
+        raise OpError("No such member in that group", status=404)
+    log.info("Member revoked: user=%s group=%s", user_id[:8], group_id[:8])
+    return {"status": "revoked", "user_id": user_id, "group_id": group_id,
+            "reminder": "rotate the group key: meshbay-node gek rotate"}
+
+
+async def unpin_member(state: dict, user_id: str) -> dict:
+    """Forget a pinned identity, so the person can pair again with a new key."""
+    roster = _roster(state)
+    if not await roster.unpin(user_id):
+        raise OpError("No such pinned identity", status=404)
+    log.info("Identity unpinned: user=%s", user_id[:8])
+    return {"status": "unpinned", "user_id": user_id}
+
+
+# ── Group keys ───────────────────────────────────────────────────────────────
+
+async def set_gek(state: dict, group_id: str, *, rotate: bool = False) -> dict:
+    """
+    Generate the group key and activate it, or rotate an existing one.
+
+    Nothing is pre-wrapped for members. Each member's copy is produced when they
+    connect, for a key they proved they hold (`join_request`) — pre-wrapping used
+    to fetch public keys from the hub, which is H3 with the node as the victim
+    instead of the inviter. Only the node's own copy is stored, so the daemon can
+    reload the key across restarts without the operator's browser.
+
+    **`rotate` generates a fresh key even when one exists.** That is the point of
+    it: after a revocation the ex-member still holds the current key, and nothing
+    else takes it away from them. Without `rotate` an existing key is kept, so
+    running this twice is not destructive by accident.
+    """
+    ctx = _group_ctx(state, group_id)
+    hub = _hub(state)
+
+    bundle_store = state.get("bundle_store")
+    if not bundle_store:
+        raise OpError("Bundle store not available", status=503)
+
+    existing = ctx.get("gek")
+    gek = generate_gek() if (rotate or not existing) else existing
+    rotated = bool(existing) and gek is not existing
+    errors: list[str] = []
+
+    roster = state.get("roster")
+    authorized = len(await roster.list_members(group_id)) if roster else 0
+
+    node_user_id = hub._session.user_id if hub._session else None
+    pk_x_node_raw = state.get("pk_x25519_raw")
+    if pk_x_node_raw and node_user_id:
+        try:
+            node_bundle = wrap_gek_aes(gek, pk_x_node_raw)
+            await bundle_store.store(
+                group_id, f"_node_{node_user_id}",
+                node_bundle["pk_eph_b64"], node_bundle["nonce_b64"],
+                node_bundle["wrapped_b64"],
+            )
+            log.info("GEK wrapped for node keystore (daemon reload)")
+        except Exception as e:
+            errors.append(f"node keystore: {e}")
+            log.warning("Failed to wrap GEK for node keystore: %s", e)
+
+    ctx["gek"] = gek
+    log.info("GEK %s for group %s — %d authorized member(s) will receive it "
+             "on connect", "rotated" if rotated else "initialized",
+             group_id[:8], authorized)
+
+    # The transport holds its own view of the group; a rotation that did not
+    # reach it would keep serving the old key until the daemon restarted.
+    for transport_key in ("webrtc", "quic_server"):
+        transport = state.get(transport_key)
+        groups = getattr(transport, "_ctx", {}).get("groups") if transport else None
+        if groups and group_id in groups:
+            groups[group_id]["gek"] = gek
+
+    indexes = state.get("indexes") or {}
+    index = indexes.get(group_id)
+    if index is not None:
+        # The index is encrypted under the GEK; leaving the old key on it would
+        # serve members a listing they cannot open.
+        index.gek = gek
+
+    return {
+        "status": "rotated" if rotated else "ok",
+        "group_id": group_id,
+        "rotated": rotated,
+        "authorized_members": authorized,
+        "errors": errors,
+    }
+
+
+# ── Groups and roots ─────────────────────────────────────────────────────────
+
+async def list_groups(state: dict) -> dict:
+    """What this node hosts, with live status. Milestone 14.2."""
+    config = state.get("config")
+    groups_ctx = state.get("groups_ctx", {})
+    peers = state.get("peers") or {}
+    out = []
+    for gid, ctx in groups_ctx.items():
+        cfg = next((g for g in config.groups if g.id == gid), None) if config else None
+        idx = ctx.get("index")
+        roots = ctx.get("roots")
+        out.append({
+            "id": gid,
+            "name": cfg.name if cfg else gid[:8],
+            "visibility": cfg.visibility if cfg else "private",
+            "join_policy": cfg.join_policy if cfg else "invite",
+            "has_gek": bool(ctx.get("gek")),
+            "file_count": idx.count if idx else 0,
+            "index_version": idx.version if idx else 0,
+            "roots": roots.describe() if roots else [],
+            "peers": sum(1 for p in peers.values() if p.get("group_id") == gid),
+        })
+    return {"groups": out}
+
+
+async def attach_group(state: dict, name: str, shared_dir: str) -> dict:
+    """
+    Write a new [[groups]] block into node.toml.
+
+    The name-to-id lookup happens here because this process is the one logged
+    into the hub. Nothing is created on the hub: the group already exists, this
+    only tells the node to host it.
+    """
+    if not name or not shared_dir:
+        raise OpError("name and shared_dir are required")
+    config = _config(state)
+    hub = _hub(state)
+    try:
+        mine = await hub.list_my_groups()
+    except Exception as e:
+        raise OpError(f"Could not list groups: {e}", status=502) from e
+
+    match = [g for g in mine if g["id"] == name or g["name"] == name]
+    if not match:
+        raise OpError(f"No group of yours is called {name!r}", status=404,
+                      extra={"available": [{"name": g["name"], "id": g["id"]}
+                                           for g in mine]})
+    if len(match) > 1:
+        raise OpError(f"Several of your groups are called {name!r} — use the id",
+                      status=409,
+                      extra={"available": [{"name": g["name"], "id": g["id"]}
+                                           for g in match]})
+    group = match[0]
+
+    if any(g.id == group["id"] for g in config.groups):
+        raise OpError(f"{group['name']!r} is already hosted by this node", status=409)
+
+    path = Path(shared_dir).expanduser()
+    try:
+        path.mkdir(parents=True, exist_ok=True)
+    except OSError as e:
+        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.
+    block = (f'\n[[groups]]\n'
+             f'id          = "{group["id"]}"\n'
+             f'name        = "{group["name"]}"\n'
+             f'visibility  = "{group.get("visibility", "private")}"\n'
+             f'\n  [[groups.roots]]\n'
+             f'  path   = "{path}"\n'
+             f'  upload = true\n')
+    try:
+        with conf_path.open("a") as f:
+            f.write(block)
+    except OSError as e:
+        raise OpError(f"Cannot write {conf_path}: {e}", status=500) from e
+
+    return {"group_id": group["id"], "name": group["name"],
+            "shared_dir": str(path), "config": str(conf_path),
+            "note": "restart the node to pick it up"}
+
+
+async def add_root(state: dict, group_id: str, path: str, *,
+                   name: str = "", kind: str = "generic",
+                   upload: bool = False) -> dict:
+    """
+    Add a directory to a group, refusing anything ambiguous.
+
+    Validated against the group's existing roots *before* being written, so a
+    config that would be refused at startup is refused here instead — where the
+    operator is watching and can fix 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)
+
+    specs = [asdict(r) for r in cfg.roots]
+    specs.append({"path": path, "name": name, "kind": kind, "upload": upload})
+    try:
+        built = RootSet.build(specs)
+    except RootError as e:
+        raise OpError(str(e)) from e
+
+    added = built.roots[-1]
+    conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
+    raise OpError(
+        # Writing into the middle of a hand-written TOML file means finding the
+        # right [[groups]] block and appending inside it, which a text append
+        # cannot do. Until that is written, say so plainly rather than appending
+        # to the wrong group.
+        f"Add this to {conf_path} under the [[groups]] block for "
+        f"{cfg.name!r}, then restart the node:\n\n"
+        f'  [[groups.roots]]\n'
+        f'  path   = "{added.path}"\n'
+        + (f'  name   = "{added.name}"\n' if name else "")
+        + (f'  kind   = "{added.kind}"\n' if kind != "generic" else "")
+        + (f'  upload = true\n' if upload else ""),
+        status=501,
+        extra={"validated": True, "name": added.name, "path": str(added.path)},
+    )
+
+
+# ── Files ────────────────────────────────────────────────────────────────────
+
+async def delete_file(state: dict, group_id: str, file_id: str) -> dict:
+    """
+    Remove a file from a group. Milestone 14.11 — the last operator action that
+    needed a browser.
+
+    Authorization happened in the adapter. On the loopback path that is the
+    session token, which means physical or SSH access to the machine hosting the
+    files — an operator who can run this can also `rm` the file, so the check is
+    not weaker than the alternative.
+    """
+    ctx = _group_ctx(state, group_id)
+    index = ctx.get("index")
+    roots = ctx.get("roots")
+    if not index or not roots:
+        raise OpError("Group has no index", status=503)
+
+    entry = index.get_entry(file_id)
+    if not entry:
+        raise OpError("No such file in this group", status=404)
+
+    from meshbay_node.roots import entry_abs_path
+    path = entry_abs_path(roots, entry)
+    if path is None:
+        raise OpError(
+            f"{entry.name!r} is in root {entry.path.split('/')[0]!r}, which is "
+            f"not readable right now — the file is frozen, not gone", status=409)
+
+    try:
+        path.unlink()
+    except FileNotFoundError:
+        # Already gone from disk; drop the stale entry rather than refusing.
+        log.warning("Index named a file that is not on disk: %s", path)
+    except OSError as e:
+        raise OpError(f"Cannot delete {entry.name!r}: {e}", status=500) from e
+
+    index.remove_entry(file_id)
+    log.info("File deleted by operator: %s/%s", entry.path, entry.name)
+    return {"status": "deleted", "name": entry.name, "path": entry.path,
+            "group_id": group_id}
+
+
+# ── Revocation denylist ──────────────────────────────────────────────────────
+
+async def read_denylist(state: dict) -> dict:
+    """Milestone 14.10 — what the node is currently refusing."""
+    denylist = state.get("denylist")
+    if not denylist:
+        return {"users": [], "groups": [], "jtis": [], "count": 0}
+    entries = denylist.entries()
+    return {**entries, "count": sum(len(v) for v in entries.values())}
+
+
+async def clear_denylist(state: dict, *, subject: str = "") -> dict:
+    """
+    Drop denylist entries — all of them, or one identifier.
+
+    Deliberately not silent: a cleared denylist re-admits whoever it was keeping
+    out, and the count is what tells the operator whether they undid one
+    revocation or all of them.
+    """
+    denylist = state.get("denylist")
+    if not denylist:
+        raise OpError("No denylist in this process", status=503)
+    removed = denylist.clear(subject)
+    log.warning("Denylist cleared (%s): %d entr(y/ies) removed",
+                subject or "all", removed)
+    return {"status": "cleared", "removed": removed, "subject": subject or "all"}
diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py
new file mode 100644
index 0000000..8f999d7
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/roots.py
@@ -0,0 +1,322 @@
+"""
+A group's content is several named roots, not one directory.
+
+    /                       (the group's virtual root)
+    ├── Films/          →   D:\\Media\\Films
+    ├── Musique/        →   E:\\Audio            (external drive)
+    └── Documents/      →   C:\\Users\\me\\Partage
+
+Every index path carries the root name as its first segment, uniformly — a group
+with one root is not a special case, because two path shapes would have to be
+kept right forever and one shape only has to be got right once.
+
+Three rules, and each of them is load-bearing rather than tidy:
+
+  * **The name is the directory's basename, derived once and stored.** Never
+    recomputed from the path, or renaming a folder on disk silently re-identifies
+    every file under it.
+  * **No root may contain another.** Otherwise the same bytes are indexed twice
+    under two identities, and deleting one leaves the other pointing at nothing.
+  * **Availability is per root.** A root whose volume goes away *freezes*: its
+    entries stay in the index, marked unavailable. Emptying it would propagate
+    deletions for a whole library as though the owner had erased it.
+
+Comparison of names is case-insensitive and NFC-normalized (`meshbay_common.paths`),
+because most of these directories live on exFAT or NTFS, where `Films` and `films`
+are one directory.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from meshbay_common.paths import fold, portable_name_problem
+
+log = logging.getLogger(__name__)
+
+VALID_KINDS = ("generic", "video", "audio", "photo")
+
+
+class RootError(ValueError):
+    """A root set that cannot be built. The message is shown to the operator."""
+
+
+@dataclass
+class Root:
+    """One named directory inside a group."""
+
+    name: str
+    path: Path
+    kind: str = "generic"
+    upload: bool = False
+    # Runtime, not configuration: set by the indexer when the directory can no
+    # longer be read, and cleared when it comes back.
+    available: bool = True
+
+    @property
+    def folded(self) -> str:
+        return fold(self.name)
+
+    def is_live(self) -> bool:
+        """Readable right now. The question `available` caches."""
+        try:
+            return self.path.is_dir()
+        except OSError:
+            return False
+
+
+def derive_name(path: Path) -> str:
+    """
+    The name a directory gets when it is added: its basename.
+
+    A path that has no usable basename — a drive root such as `E:\\`, or `/` —
+    has nothing to derive from, and the operator has to supply a name.
+    """
+    name = path.name or ""
+    if not name:
+        raise RootError(
+            f"{path} has no directory name to use — give the root an explicit "
+            f"name (a drive or filesystem root cannot supply one)")
+    return name
+
+
+@dataclass
+class RootSet:
+    """
+    The roots of one group, and the only place a virtual path is resolved.
+
+    `resolve()` is the single entry point for turning something that arrived
+    over the wire into a path on disk. Callers must not join paths themselves —
+    that is how a traversal gets in through the one site nobody reviewed.
+    """
+
+    roots: list[Root] = field(default_factory=list)
+
+    # ── Construction ─────────────────────────────────────────────────────────
+
+    @classmethod
+    def build(cls, specs: list[dict]) -> "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.
+        """
+        roots: list[Root] = []
+        by_folded: dict[str, Root] = {}
+
+        for spec in specs:
+            raw = str(spec.get("path", "")).strip()
+            if not raw:
+                raise RootError("a root has no path")
+            path = Path(raw).expanduser()
+            try:
+                path = path.resolve()
+            except OSError as e:
+                raise RootError(f"{raw}: {e}") from e
+
+            name = str(spec.get("name") or "").strip() or derive_name(path)
+
+            problem = portable_name_problem(name)
+            if problem:
+                raise RootError(
+                    f"root name {name!r} ({problem}) — every member sees this as a "
+                    f"folder name, including on Windows. Give the root an explicit "
+                    f"name in the config")
+
+            clash = by_folded.get(fold(name))
+            if clash:
+                raise RootError(
+                    f"two roots would both be called {name!r}: {clash.path} and "
+                    f"{path}. Names are compared without regard to case. Give one "
+                    f"of them an explicit name")
+
+            kind = str(spec.get("kind") or "generic").strip().lower()
+            if kind not in VALID_KINDS:
+                log.warning("root %r: unknown kind %r — using 'generic'", name, kind)
+                kind = "generic"
+
+            root = Root(name=name, path=path, kind=kind,
+                        upload=bool(spec.get("upload", 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:
+        target = fold(name)
+        for root in self.roots:
+            if root.folded == target:
+                return root
+        return None
+
+    @property
+    def upload_root(self) -> Root | None:
+        for root in self.roots:
+            if root.upload:
+                return root
+        return None
+
+    @property
+    def names(self) -> list[str]:
+        return [r.name for r in self.roots]
+
+    def __bool__(self) -> bool:
+        return bool(self.roots)
+
+    def __len__(self) -> int:
+        return len(self.roots)
+
+    def __iter__(self):
+        return iter(self.roots)
+
+    # ── Resolution ───────────────────────────────────────────────────────────
+
+    def split(self, virtual: str) -> tuple[Root, str] | None:
+        """
+        `"Films/2024"` → (the Films root, `"2024"`). None if no such root.
+
+        Does not touch the filesystem, so it is safe to call on a root whose
+        volume is absent.
+        """
+        rel = (virtual or "").strip().strip("/")
+        if not rel:
+            return None
+        head, _, tail = rel.partition("/")
+        root = self.by_name(head)
+        if root is None:
+            return None
+        return root, tail
+
+    def resolve(self, virtual: str, *, require_available: bool = True) -> Path | None:
+        """
+        A virtual path from the wire → a real path inside its root, or None.
+
+        Refuses `..`, absolute segments, and anything whose resolved form escapes
+        the root — symlinks included, which is why this resolves before
+        comparing rather than checking the string.
+        """
+        found = self.split(virtual)
+        if found is None:
+            return None
+        root, tail = found
+        if require_available and not root.available:
+            return None
+
+        parts = [seg for seg in tail.split("/") if seg not in ("", ".")]
+        if any(seg == ".." for seg in parts):
+            return None
+        try:
+            target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve()
+            base = root.path.resolve()
+        except OSError:
+            return None
+        if target != base and base not in target.parents:
+            return None
+        return target
+
+    def virtual_of(self, absolute: Path) -> str | None:
+        """A real path anywhere in this group → `"Films/2024"`, or None."""
+        for root in self.roots:
+            virtual = self.virtual_path(root, absolute)
+            if virtual is not None:
+                return virtual
+        return None
+
+    def virtual_path(self, root: Root, absolute: Path) -> str | None:
+        """The inverse of `resolve`: a real path → `"Films/2024"`."""
+        try:
+            rel = absolute.resolve().relative_to(root.path.resolve())
+        except (ValueError, OSError):
+            return None
+        return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"
+
+    # ── Availability ─────────────────────────────────────────────────────────
+
+    def refresh_availability(self) -> list[tuple[Root, bool]]:
+        """
+        Re-read which roots are readable. Returns the ones that changed.
+
+        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.
+        """
+        changed: list[tuple[Root, bool]] = []
+        for root in self.roots:
+            live = root.is_live()
+            if live != root.available:
+                root.available = live
+                changed.append((root, live))
+                log.warning("Root %r is now %s (%s)", root.name,
+                            "available" if live else "UNAVAILABLE", root.path)
+        return changed
+
+    def describe(self) -> list[dict]:
+        """Per-root state for the index payload and the admin UI."""
+        return [
+            {"name": r.name, "kind": r.kind, "available": r.available,
+             "upload": r.upload}
+            for r in self.roots
+        ]
+
+
+def entry_abs_path(roots: RootSet, entry) -> Path | None:
+    """
+    Where an index entry actually lives, or None if its root is gone.
+
+    The one place an entry becomes a path. `entry.path` starts with a root name,
+    so a group whose drive is unplugged answers None here rather than opening
+    something unrelated that happens to share a relative path with another root.
+    """
+    parent = roots.resolve(entry.path)
+    return (parent / entry.name) if parent else None
+
+
+def _refuse_nesting(new: Root, existing: list[Root]) -> None:
+    """
+    No root may contain another, compared case-insensitively.
+
+    `D:\\Media` and `D:\\Media\\Films` together would index the same bytes twice
+    under two identities. On NTFS and exFAT `d:\\media` is the same directory as
+    `D:\\Media`, so a string comparison that respects case would miss it.
+    """
+    new_parts = [fold(p) for p in new.path.parts]
+    for other in existing:
+        other_parts = [fold(p) for p in other.path.parts]
+        if new_parts == other_parts:
+            raise RootError(
+                f"roots {new.name!r} and {other.name!r} are the same directory "
+                f"({new.path})")
+        shorter, longer, inner, outer = (
+            (other_parts, new_parts, new, other)
+            if len(other_parts) < len(new_parts)
+            else (new_parts, other_parts, other, new))
+        if longer[:len(shorter)] == shorter:
+            raise RootError(
+                f"root {inner.name!r} ({inner.path}) is inside root "
+                f"{outer.name!r} ({outer.path}) — its files would be indexed "
+                f"twice under two names")
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
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index b78f78d..050829e 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -23,6 +23,7 @@ from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
 from fastapi.responses import HTMLResponse, JSONResponse
 
 from meshbay_node import __version__
+from meshbay_node import ops
 from meshbay_node.config import DEFAULT_CONFIG_PATH
 from meshbay_common.crypto import generate_gek, wrap_gek_aes
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
@@ -30,6 +31,24 @@ from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
 log = logging.getLogger(__name__)
 
 
+def _op(coro):
+    """
+    Run an operation and translate its refusal into a JSON response.
+
+    The operations live in `meshbay_node.ops` and know nothing about HTTP. This
+    is the whole of the HTTP adapter: without it each handler would carry its
+    own status codes, and the MNP handler in Stage B3 would carry a second set
+    that slowly stopped agreeing.
+    """
+    async def run():
+        try:
+            return await coro()
+        except ops.OpError as e:
+            return JSONResponse(e.as_dict(), e.status)
+    return run()
+
+
+
 def create_ui_app(state: dict) -> FastAPI:
     app = FastAPI(
         title="MeshBay Node Admin",
@@ -110,90 +129,27 @@ def create_ui_app(state: dict) -> FastAPI:
 
     @app.get("/api/groups")
     async def api_groups():
-        groups_ctx = state.get("groups_ctx", {})
-        config = state.get("config")
-        result = []
-        for gid, ctx in groups_ctx.items():
-            cfg = None
-            if config:
-                cfg = next((g for g in config.groups if g.id == gid), None)
-            idx = ctx.get("index")
-            result.append({
-                "id": gid,
-                "name": cfg.name if cfg else gid[:8],
-                "shared_dir": str(ctx.get("shared_root", "")),
-                "visibility": cfg.visibility if cfg else "private",
-                "file_count": idx.count if idx else 0,
-                "index_version": idx.version if idx else 0,
-            })
-        return {"groups": result}
-
+        return await _op(lambda: ops.list_groups(state))
     @app.post("/api/groups/attach")
     async def attach_group(payload: dict):
-        """
-        Write a new [[groups]] block into node.toml.
+        return await _op(lambda: ops.attach_group(
+            state,
+            (payload.get("name") or "").strip(),
+            (payload.get("shared_dir") or "").strip(),
+        ))
 
-        The name-to-id lookup happens here because this process is the one logged
-        into the hub. Nothing is created on the hub: the group already exists,
-        this only tells the node to host it.
-        """
-        name = (payload.get("name") or "").strip()
-        shared_dir = (payload.get("shared_dir") or "").strip()
-        if not name or not shared_dir:
-            return JSONResponse({"error": "name and shared_dir are required"}, 400)
+    @app.delete("/api/groups/{group_id}/files/{file_id}")
+    async def delete_file(group_id: str, file_id: str):
+        """Milestone 14.11 — the last operator action that needed a browser."""
+        return await _op(lambda: ops.delete_file(state, group_id, file_id))
 
-        config = state.get("config")
-        if not config:
-            return JSONResponse({"error": "No config loaded"}, 503)
+    @app.get("/api/denylist")
+    async def api_denylist():
+        return await _op(lambda: ops.read_denylist(state))
 
-        hub = state.get("hub")
-        if not hub or not hub._session:
-            return JSONResponse({"error": "Hub not connected"}, 503)
-        try:
-            mine = await hub.list_my_groups()
-        except Exception as e:
-            return JSONResponse({"error": f"Could not list groups: {e}"}, 502)
-
-        match = [g for g in mine if g["id"] == name or g["name"] == name]
-        if not match:
-            return JSONResponse({
-                "error": f"No group of yours is called {name!r}",
-                "available": [{"name": g["name"], "id": g["id"]} for g in mine],
-            }, 404)
-        if len(match) > 1:
-            return JSONResponse({
-                "error": f"Several of your groups are called {name!r} — use the id",
-                "available": [{"name": g["name"], "id": g["id"]} for g in match],
-            }, 409)
-        group = match[0]
-
-        if any(g.id == group["id"] for g in config.groups):
-            return JSONResponse(
-                {"error": f"{group['name']!r} is already hosted by this node"}, 409)
-
-        path = Path(shared_dir).expanduser()
-        try:
-            path.mkdir(parents=True, exist_ok=True)
-        except OSError as e:
-            return JSONResponse({"error": f"Cannot create {path}: {e}"}, 400)
-
-        # 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.
-        conf_path = Path(state.get("config_path") or DEFAULT_CONFIG_PATH)
-        block = (f'\n[[groups]]\n'
-                 f'id          = "{group["id"]}"\n'
-                 f'name        = "{group["name"]}"\n'
-                 f'shared_dir  = "{path}"\n'
-                 f'visibility  = "{group.get("visibility", "private")}"\n')
-        try:
-            with conf_path.open("a") as f:
-                f.write(block)
-        except OSError as e:
-            return JSONResponse({"error": f"Cannot write {conf_path}: {e}"}, 500)
-
-        return {"group_id": group["id"], "name": group["name"],
-                "shared_dir": str(path), "config": str(conf_path)}
+    @app.post("/api/denylist/clear")
+    async def api_denylist_clear(subject: str = ""):
+        return await _op(lambda: ops.clear_denylist(state, subject=subject))
 
     @app.get("/api/groups/{group_id}/files")
     async def api_group_files(group_id: str):
@@ -279,7 +235,11 @@ def create_ui_app(state: dict) -> FastAPI:
                 {
                     "id": g.id,
                     "name": g.name,
-                    "shared_dir": g.shared_dir,
+                    "roots": [
+                        {"path": r.path, "name": r.name, "kind": r.kind,
+                         "upload": r.upload}
+                        for r in g.roots
+                    ],
                     "visibility": g.visibility,
                 }
                 for g in config.groups
@@ -290,205 +250,33 @@ def create_ui_app(state: dict) -> FastAPI:
 
     @app.post("/api/operator/pair")
     async def operator_pair():
-        """
-        Issue a one-time code that pairs a browser as this node's operator.
-
-        The code is the whole point: it binds the operator's browser identity key
-        to their account without asking the hub, which is what stops a hub from
-        naming itself node administrator (M3, and the same substitution as H3).
-        It is returned once and stored only as a hash.
-        """
-        roster = state.get("roster")
-        user_id = state.get("node_user_id")
-        if not roster or not user_id:
-            return JSONResponse({"error": "Node not connected to hub yet"}, 503)
-
-        config = state.get("config")
-        ttl = (config.node.pair_ttl_hours if config else 24) * 3600
-        code = await roster.create_invite(
-            group_id="",                 # operator authority is node-wide
-            user_id=user_id,
-            role=ROLE_OPERATOR,
-            created_by="local-cli",
-            ttl=ttl,
-            username=(config.hub.username if config else ""),
-        )
-        invites = await roster.list_invites()
-        expires = next((i["expires_at"] for i in invites
-                        if i["user_id"] == user_id and i["role"] == ROLE_OPERATOR), "")
-        return {"code": code, "expires_at": expires, "user_id": user_id}
+        return await _op(lambda: ops.pair_operator(state))
 
     @app.get("/api/roster")
     async def api_roster(group_id: str = ""):
-        roster = state.get("roster")
-        if not roster:
-            return {"identities": [], "members": [], "invites": []}
-        return {
-            "identities": await roster.list_identities(),
-            "members": await roster.list_members(group_id or None),
-            "invites": await roster.list_invites(),
-        }
+        return await _op(lambda: ops.read_roster(state, group_id))
 
     @app.post("/api/groups/{group_id}/invites")
     async def create_invite(group_id: str, username: str):
-        """
-        Issue an invitation code from the CLI, without a browser.
-
-        The hub is asked for the account id and nothing else — never for a key.
-        A hub that answered with the wrong account would produce an invite whose
-        code it never learns, since the code goes to a human out of band.
-        """
-        roster = state.get("roster")
-        groups_ctx = state.get("groups_ctx", {})
-        if not roster:
-            return JSONResponse({"error": "Roster not available"}, 503)
-        if group_id not in groups_ctx:
-            return JSONResponse({"error": "Group not hosted on this node"}, 404)
-
-        hub = state.get("hub")
-        if not hub or not hub._session:
-            return JSONResponse({"error": "Hub not connected"}, 503)
-        try:
-            account = await hub.get_user_pubkeys(username)
-        except Exception as e:
-            return JSONResponse({"error": f"Unknown user {username!r}: {e}"}, 404)
-
-        config = state.get("config")
-        ttl = (config.node.invite_ttl_hours if config else 168) * 3600
-        code = await roster.create_invite(
-            group_id=group_id,
-            user_id=account["user_id"],
-            role=ROLE_MEMBER,
-            created_by="local-cli",
-            ttl=ttl,
-            username=username,
-        )
-        invites = await roster.list_invites()
-        expires = next((i["expires_at"] for i in invites
-                        if i["user_id"] == account["user_id"]
-                        and i["group_id"] == group_id), "")
-        return {"code": code, "expires_at": expires,
-                "username": username, "user_id": account["user_id"]}
+        return await _op(lambda: ops.create_invite(state, group_id, username))
 
     @app.get("/api/resolve")
     async def resolve_user(username: str):
-        """
-        Map a username to an account id for the CLI.
-
-        The roster answers first — it is the node's own record. The hub is the
-        fallback for identities pinned before invitations carried a name, and for
-        people admitted through an open-join group. Only an account id comes back;
-        no key is ever taken from here.
-        """
-        roster = state.get("roster")
-        if roster:
-            for ident in await roster.list_identities():
-                if ident["username"] == username:
-                    return {"user_id": ident["user_id"], "source": "roster"}
-        hub = state.get("hub")
-        if hub and hub._session:
-            try:
-                account = await hub.get_user_pubkeys(username)
-                return {"user_id": account["user_id"], "source": "hub"}
-            except Exception:
-                pass
-        return JSONResponse({"error": f"Unknown user {username!r}"}, 404)
+        return await _op(lambda: ops.resolve_user(state, username))
 
     @app.post("/api/members/{user_id}/revoke")
     async def revoke_member(user_id: str, group_id: str):
-        """
-        Stop serving the group key to someone.
-
-        Takes effect on their next connection: the key is wrapped on demand, so
-        there is no stored bundle left behind that would outlive this. Rotating
-        the group key is still required — they hold the current one.
-        """
-        roster = state.get("roster")
-        if not roster:
-            return JSONResponse({"error": "Roster not available"}, 503)
-        if not await roster.set_status(group_id, user_id, "revoked"):
-            return JSONResponse({"error": "No such member in that group"}, 404)
-        log.info("Member revoked: user=%s group=%s", user_id[:8], group_id[:8])
-        return {"status": "revoked", "user_id": user_id, "group_id": group_id,
-                "reminder": "rotate the group key: meshbay-node gek-init"}
+        return await _op(lambda: ops.revoke_member(state, user_id, group_id))
 
     @app.post("/api/members/{user_id}/unpin")
     async def unpin_member(user_id: str):
-        """Forget a pinned identity, so the person can pair again with a new key."""
-        roster = state.get("roster")
-        if not roster:
-            return JSONResponse({"error": "Roster not available"}, 503)
-        if not await roster.unpin(user_id):
-            return JSONResponse({"error": "No such pinned identity"}, 404)
-        log.info("Identity unpinned: user=%s", user_id[:8])
-        return {"status": "unpinned", "user_id": user_id}
+        return await _op(lambda: ops.unpin_member(state, user_id))
 
     # ── GEK initialization (operator only, localhost) ──────────────────────
 
     @app.post("/api/groups/{group_id}/gek")
-    async def init_gek(group_id: str):
-        """
-        Generate the group key and activate it.
-
-        It used to be wrapped here for every member, using public keys fetched from
-        the hub — which is H3 with the node as the victim instead of the inviter: a
-        hub answering with its own key was handed the group key by the node itself.
-
-        Nothing is pre-wrapped for members now. Each member's copy is produced when
-        they connect, for a key they proved they hold (`join_request`). Only the
-        node's own copy is stored, so the daemon can reload the key across restarts
-        without the operator's browser.
-        """
-        groups_ctx = state.get("groups_ctx", {})
-        if group_id not in groups_ctx:
-            return JSONResponse({"error": "Group not hosted on this node"}, 404)
-
-        hub = state.get("hub")
-        if not hub or not hub._session:
-            return JSONResponse({"error": "Hub not connected"}, 503)
-
-        bundle_store = state.get("bundle_store")
-        if not bundle_store:
-            return JSONResponse({"error": "Bundle store not available"}, 503)
-
-        existing_gek = groups_ctx[group_id].get("gek")
-        gek = existing_gek or generate_gek()
-        errors: list[str] = []
-
-        roster = state.get("roster")
-        authorized = len(await roster.list_members(group_id)) if roster else 0
-
-        # Store a copy wrapped for the node keystore X25519 key so the daemon can
-        # reload the GEK on restart without the operator's browser keys.
-        node_user_id = hub._session.user_id if hub._session else None
-        pk_x_node_raw = state.get("pk_x25519_raw")
-        if pk_x_node_raw and node_user_id:
-            try:
-                node_bundle = wrap_gek_aes(gek, pk_x_node_raw)
-                await bundle_store.store(
-                    group_id, f"_node_{node_user_id}",
-                    node_bundle["pk_eph_b64"], node_bundle["nonce_b64"],
-                    node_bundle["wrapped_b64"],
-                )
-                log.info("GEK wrapped for node keystore (daemon reload)")
-            except Exception as e:
-                errors.append(f"node keystore: {e}")
-                log.warning("Failed to wrap GEK for node keystore: %s", e)
-
-        groups_ctx[group_id]["gek"] = gek
-        log.info("GEK initialized for group %s — %d authorized member(s) will "
-                 "receive it on connect", group_id[:8], authorized)
-
-        webrtc = state.get("webrtc")
-        if webrtc and "groups" in webrtc._ctx and group_id in webrtc._ctx["groups"]:
-            webrtc._ctx["groups"][group_id]["gek"] = gek
-
-        return {
-            "status": "ok",
-            "group_id": group_id,
-            "authorized_members": authorized,
-            "errors": errors,
-        }
+    async def init_gek(group_id: str, rotate: bool = False):
+        return await _op(lambda: ops.set_gek(state, group_id, rotate=rotate))
 
     # ── Chat endpoints ───────────────────────────────────────────────────────
 
@@ -656,7 +444,15 @@ def _render_page(state: dict, roster_view: dict | None = None) -> str:
             cfg = next((g for g in config.groups if g.id == gid), None)
         idx = ctx.get("index")
         name = cfg.name if cfg else gid[:8]
-        shared = ctx.get("shared_root", "")
+        roots = ctx.get("roots")
+        # An unavailable root is shown as such rather than hidden: its files are
+        # still listed and still in the index, and hiding the root would make a
+        # frozen library look deleted — the exact confusion this is meant to
+        # prevent.
+        shared = ", ".join(
+            f"{r.name} → {r.path}" + ("" if r.available else "  [UNAVAILABLE]")
+            for r in roots
+        ) if roots else ""
         vis = cfg.visibility if cfg else "private"
         fcount = idx.count if idx else 0
         total_size = sum(e.size for e in idx.entries) if idx else 0