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>

40 files changed, 3489 insertions, 521 deletions

diff --git a/packages/meshbay-common/src/meshbay_common/__init__.py b/packages/meshbay-common/src/meshbay_common/__init__.py
index 60a9dc7..814c606 100644
--- a/packages/meshbay-common/src/meshbay_common/__init__.py
+++ b/packages/meshbay-common/src/meshbay_common/__init__.py
@@ -4,5 +4,5 @@ __version__ = "0.5.0"
 # 0.2: added PING/PONG, and `before`/`has_more` on chat history. Both are
 # additive — an 0.1 peer sends no `before` and gets the newest page, which is
 # what it wanted — so this is a MINOR bump, not a MAJOR one.
-MNP_VERSION = "0.2"
+MNP_VERSION = "0.3"
 MHP_VERSION = "0.1"
diff --git a/packages/meshbay-common/src/meshbay_common/adminop.py b/packages/meshbay-common/src/meshbay_common/adminop.py
index fe4be83..4e28f65 100644
--- a/packages/meshbay-common/src/meshbay_common/adminop.py
+++ b/packages/meshbay-common/src/meshbay_common/adminop.py
@@ -37,6 +37,14 @@ OP_FILE_DELETE = "file_delete"
 OP_DIR_DELETE = "dir_delete"
 OP_INVITE_CREATE = "invite_create"
 OP_MEMBER_REVOKE = "member_revoke"
+# Rotating the group key is what actually takes it away from a revoked member:
+# revocation stops the node serving the *next* key, and they still hold the
+# current one. The node generates the new key itself with its own CSPRNG, so
+# nothing arriving over MNP contributes key material — the C5b rule is about
+# key material from outside, not about the instruction.
+OP_GEK_ROTATE = "gek_rotate"
+# Forgetting a pinned identity, so someone can pair again after losing a device.
+OP_MEMBER_UNPIN = "member_unpin"
 # OP_GEK_BUNDLE_STORE is gone. Members no longer hand the node key material at
 # all: the node holds the GEK and wraps it itself, for a key the recipient proved
 # they hold (see `join.py` and docs/invite-pairing-v1.md). The operation existed
diff --git a/packages/meshbay-common/src/meshbay_common/paths.py b/packages/meshbay-common/src/meshbay_common/paths.py
new file mode 100644
index 0000000..45bf34e
--- /dev/null
+++ b/packages/meshbay-common/src/meshbay_common/paths.py
@@ -0,0 +1,150 @@
+"""
+Name rules that have to give the same answer everywhere.
+
+Most people share from an external drive formatted exFAT or NTFS, on Windows.
+Those filesystems are case-insensitive and case-preserving, store no POSIX
+permissions and no symlinks, and Windows refuses a set of names outright. So two
+names that are plainly different on ext4 can be the same file somewhere else, or
+no file at all — and an index built on one machine is read on another.
+
+"Are these two names the same?" therefore has one answer, and it lives here.
+
+What this module deliberately does **not** do: rewrite names. A name is stored as
+the filesystem gave it, because that is the string that opens the file. Folding
+and normalization exist for *comparison*, never for storage.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import unicodedata
+from pathlib import Path, PurePath
+
+# Windows refuses these as a basename, with or without an extension: `AUX.txt`
+# is as impossible as `AUX`. A group indexed on Linux can hold them, and a
+# Windows client then cannot write the file it just downloaded.
+WINDOWS_RESERVED = frozenset({
+    "CON", "PRN", "AUX", "NUL",
+    *(f"COM{i}" for i in range(1, 10)),
+    *(f"LPT{i}" for i in range(1, 10)),
+})
+
+# Reserved on Windows; `/` is reserved everywhere. Control characters go too.
+_RESERVED_CHARS = set('<>:"/\\|?*') | {chr(c) for c in range(32)}
+
+# Windows without long-path support. A deep media library reaches this.
+MAX_PATH_WINDOWS = 260
+
+
+def nfc(text: str) -> str:
+    """
+    Canonical form for comparison.
+
+    `Café.mkv` written on macOS (NFD: `e` + combining acute) and on Windows
+    (NFC: precomposed `é`) are different byte strings that name the same file to
+    a human, and to most filesystems. For French names this is routine.
+    """
+    return unicodedata.normalize("NFC", text)
+
+
+def fold(text: str) -> str:
+    """
+    The form in which two names are "the same file".
+
+    NFC first, then `casefold` — which is not `lower()`: it handles the cases
+    `lower()` misses, and those are the ones that show up as a bug report rather
+    than a test failure.
+    """
+    return nfc(text).casefold()
+
+
+def fold_path(rel: str) -> str:
+    """`fold` applied per segment, so separators survive."""
+    return "/".join(fold(seg) for seg in rel.split("/"))
+
+
+def portable_name_problem(name: str) -> str | None:
+    """
+    Why `name` cannot be written on some supported platform, or None.
+
+    Used to warn, not to refuse: a file already on the operator's disk is a fact,
+    and the answer is to tell whoever downloads it that it was renamed — not to
+    pretend it is not there.
+    """
+    if not name:
+        return "empty name"
+    if name in (".", ".."):
+        return "reserved name"
+    bad = sorted(set(name) & _RESERVED_CHARS)
+    if bad:
+        printable = "".join(c if c.isprintable() else "?" for c in bad)
+        return f"contains reserved characters: {printable}"
+    # Windows strips these silently, so `file .` and `file` become the same
+    # thing after a round trip.
+    if name[-1] in " .":
+        return "ends with a space or a dot"
+    stem = name.split(".", 1)[0]
+    if stem.upper() in WINDOWS_RESERVED:
+        return f"reserved on Windows ({stem.upper()})"
+    return None
+
+
+def is_portable_name(name: str) -> bool:
+    return portable_name_problem(name) is None
+
+
+def long_path(path: Path | str) -> str:
+    """
+    A path string safe to hand to the OS.
+
+    On Windows, prefix with `\\\\?\\` so `MAX_PATH` does not truncate a deep
+    library. The prefix requires an absolute, already-normalized path, and it is
+    a no-op everywhere else.
+    """
+    text = str(path)
+    if os.name != "nt" or text.startswith("\\\\?\\"):
+        return text
+    resolved = str(PurePath(text))
+    if len(resolved) < MAX_PATH_WINDOWS and not text.startswith("\\\\"):
+        return text
+    if text.startswith("\\\\"):          # UNC: \\server\share → \\?\UNC\server\share
+        return "\\\\?\\UNC" + text[1:]
+    return "\\\\?\\" + text
+
+
+def find_fold_collisions(names: list[str]) -> dict[str, list[str]]:
+    """
+    Names that collide once folded, keyed by the folded form.
+
+    Two entries that fold together cannot both exist on a case-insensitive
+    filesystem. On ext4 they can, which is how an index becomes unrepresentable
+    for the person who downloads it — so this is reported to the operator rather
+    than resolved silently: only they know which file they meant.
+    """
+    seen: dict[str, list[str]] = {}
+    for name in names:
+        seen.setdefault(fold(name), []).append(name)
+    return {k: v for k, v in seen.items() if len(v) > 1}
+
+
+_TRAILING = re.compile(r"[ .]+$")
+
+
+def sanitize_for_download(name: str, *, replacement: str = "_") -> str:
+    """
+    A name that can be written on the running platform, from one that may not be.
+
+    For the client saving a file, never for the node storing one. Returns the
+    name unchanged when it is already portable, so the common case is identity
+    and the caller can tell whether it renamed anything by comparing.
+    """
+    if is_portable_name(name):
+        return name
+    out = "".join(replacement if c in _RESERVED_CHARS else c for c in name)
+    out = _TRAILING.sub("", out)
+    stem, dot, ext = out.partition(".")
+    if stem.upper() in WINDOWS_RESERVED:
+        stem = f"{stem}{replacement}"
+    out = f"{stem}{dot}{ext}"
+    return out or "unnamed"
diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py
index 1aadafe..a1c971f 100644
--- a/packages/meshbay-common/src/meshbay_common/protocol.py
+++ b/packages/meshbay-common/src/meshbay_common/protocol.py
@@ -11,8 +11,10 @@ Every message carries a "v" field for protocol version.
 from dataclasses import dataclass, field
 from typing import Any
 
-MNP_VERSION = "0.1"
-MHP_VERSION = "0.1"
+# The wire versions live in meshbay_common/__init__.py — one source, because a
+# second copy here said "0.1" while every message on the wire carried "0.2".
+# Nothing imported it, which is the only reason it was harmless.
+from meshbay_common import MNP_VERSION, MHP_VERSION   # noqa: F401  (re-export)
 
 
 # ── MNP message types ─────────────────────────────────────────────────────────
@@ -74,6 +76,12 @@ class MNP:
     INVITE_CREATE        = "invite_create"         # operator → node: issue a pairing code
     MEMBER_REVOKE        = "member_revoke"         # operator → node: stop serving the key
     MEMBER_REVOKE_ACK    = "member_revoke_ack"
+    MEMBER_UNPIN         = "member_unpin"          # operator → node: forget an identity
+    MEMBER_UNPIN_ACK     = "member_unpin_ack"
+    # Rotation is the half of revocation that revocation cannot do: the node
+    # generates a fresh key itself, so no key material crosses the wire.
+    GEK_ROTATE           = "gek_rotate"            # operator → node: new group key
+    GEK_ROTATE_ACK       = "gek_rotate_ack"
     INVITE_RESULT        = "invite_result"         # node → operator: the code, once
 
 
diff --git a/packages/meshbay-common/tests/test_paths.py b/packages/meshbay-common/tests/test_paths.py
new file mode 100644
index 0000000..b9052e3
--- /dev/null
+++ b/packages/meshbay-common/tests/test_paths.py
@@ -0,0 +1,123 @@
+"""
+Name rules that have to give the same answer on every platform.
+
+Most people share from an external drive formatted exFAT or NTFS, on Windows, so
+these are not compatibility niceties — they decide whether two index entries are
+one file, and whether a member can save what they downloaded.
+"""
+
+import pytest
+
+from meshbay_common.paths import (
+    find_fold_collisions,
+    fold,
+    fold_path,
+    is_portable_name,
+    nfc,
+    portable_name_problem,
+    sanitize_for_download,
+)
+
+
+# ── Same file or not ─────────────────────────────────────────────────────────
+
+def test_case_differences_fold_together():
+    assert fold("README.TXT") == fold("readme.txt")
+
+
+def test_folding_is_not_lowercasing():
+    """`casefold` handles what `lower()` misses, and those are the pairs that
+    arrive as a bug report rather than a test failure."""
+    assert fold("STRASSE") == fold("straße")
+    assert "STRASSE".lower() != "straße".lower()
+
+
+def test_the_two_unicode_spellings_of_an_accent_are_one_name():
+    """
+    `Café.mkv` written on macOS (NFD) and on Windows (NFC) are different byte
+    strings naming the same file. For French filenames this is routine.
+    """
+    nfc_form = "Café.mkv"          # é precomposed
+    nfd_form = "Café.mkv"         # e + combining acute
+    assert nfc_form != nfd_form
+    assert nfc(nfd_form) == nfc_form
+    assert fold(nfd_form) == fold(nfc_form)
+
+
+def test_distinct_names_stay_distinct():
+    assert fold("film.mkv") != fold("film2.mkv")
+
+
+def test_folding_a_path_keeps_its_separators():
+    assert fold_path("Films/2024/A.MKV") == "films/2024/a.mkv"
+
+
+def test_collisions_are_found_and_grouped():
+    found = find_fold_collisions(
+        ["README.txt", "readme.TXT", "notes.md", "Café.mkv", "Café.mkv"])
+    groups = {frozenset(v) for v in found.values()}
+    assert len(groups) == 2
+    assert frozenset({"README.txt", "readme.TXT"}) in groups
+    # The other pair is the two Unicode spellings of the same accented name,
+    # which are different byte strings — so compare by size, not by literal.
+    assert any(len(g) == 2 and all(f.endswith('.mkv') for f in g)
+               for g in groups)
+    assert all("notes.md" not in v for v in found.values())
+
+
+def test_no_collision_means_no_report():
+    assert find_fold_collisions(["a.txt", "b.txt"]) == {}
+
+
+# ── What Windows refuses ─────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("name", ["CON", "PRN", "AUX", "NUL", "COM1", "LPT9",
+                                  "aux", "Aux.txt", "COM3.tar.gz"])
+def test_windows_reserved_names_are_flagged(name):
+    """`AUX.txt` is as impossible as `AUX` — the extension does not help."""
+    assert portable_name_problem(name) is not None
+
+
+@pytest.mark.parametrize("name", ['a<b', 'a>b', 'a:b', 'a"b', "a/b", "a\\b",
+                                  "a|b", "a?b", "a*b", "a\x00b", "a\tb"])
+def test_reserved_characters_are_flagged(name):
+    assert portable_name_problem(name) is not None
+
+
+@pytest.mark.parametrize("name", ["trailing ", "trailing.", "  "])
+def test_a_trailing_space_or_dot_is_flagged(name):
+    """Windows strips them silently, so `file .` and `file` come back the same."""
+    assert portable_name_problem(name) is not None
+
+
+@pytest.mark.parametrize("name", ["film.mkv", "Café.mkv", "rapport (1).pdf",
+                                  "été.txt", "COMET.txt", "AUXILIARY.doc",
+                                  "日本語.txt"])
+def test_ordinary_names_are_portable(name):
+    assert is_portable_name(name), portable_name_problem(name)
+
+
+def test_empty_and_dot_names_are_refused():
+    assert portable_name_problem("") is not None
+    assert portable_name_problem(".") is not None
+    assert portable_name_problem("..") is not None
+
+
+# ── Saving a file whose name the platform refuses ────────────────────────────
+
+def test_a_portable_name_is_returned_unchanged():
+    """Identity in the common case, so a caller can tell whether it renamed
+    anything by comparing."""
+    assert sanitize_for_download("film.mkv") == "film.mkv"
+    assert sanitize_for_download("Café (2024).mkv") == "Café (2024).mkv"
+
+
+def test_sanitizing_produces_something_writable():
+    for hostile in ["a<b>c.txt", "AUX", "trailing .", "with/slash.txt", "COM1.log"]:
+        cleaned = sanitize_for_download(hostile)
+        assert is_portable_name(cleaned), f"{hostile!r} → {cleaned!r} still refused"
+
+
+def test_sanitizing_never_returns_nothing():
+    assert sanitize_for_download("...") not in ("", None)
+    assert sanitize_for_download("???") not in ("", None)
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js
index cb7a462..4a4e712 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/app.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js
@@ -1240,6 +1240,10 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth,
   // Directories are not index entries, so a new empty one needs a nudge
   // to appear in the breadcrumb listing.
   const [nodeDirs, setNodeDirs] = useState([]);
+  // The group's roots and whether each is readable. A root whose drive is
+  // unplugged keeps its files listed — they are frozen, not deleted — so this is
+  // the only thing that lets the UI say which of the two it is.
+  const [nodeRoots, setNodeRoots] = useState([]);
 
   const [isNodeAdmin, setIsNodeAdmin] = useState(false);
   // Paired ≠ operator account. `is_node_admin` says the hub account owning this
@@ -1275,6 +1279,7 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth,
     const fresh = indexMsg.entries || [];
     setEntries(fresh);
     if (indexMsg.dirs) setNodeDirs(indexMsg.dirs);
+      if (indexMsg.roots) setNodeRoots(indexMsg.roots);
     cacheGroupIndex(groupId, group ? group.name : groupId, fresh);
   }, [groupId, group]);
 
@@ -1492,6 +1497,7 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth,
       const indexMsg = await transport.fetchIndex();
       if (indexMsg.entries) setEntries(indexMsg.entries);
       if (indexMsg.dirs) setNodeDirs(indexMsg.dirs);
+      if (indexMsg.roots) setNodeRoots(indexMsg.roots);
     } catch (err) {
       setError(err.message);
     }
@@ -1671,6 +1677,19 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth,
   }
   const subdirs = [...dirs].sort();
 
+  // At the top of a group the folders on screen ARE the roots, so their state
+  // belongs there. Deeper in, everything shown lives inside one readable root
+  // and there is nothing to flag.
+  const rootState = new Map(nodeRoots.map(r => [r.name, r]));
+  const unavailableHere = currentPath
+    ? []
+    : subdirs.filter(d => rootState.get(d) && rootState.get(d).available === false);
+  // A member cannot create a folder at the top of a group: that level is the
+  // set of roots, which is the operator's configuration and not a directory on
+  // anyone's disk. The node refuses it, so offering it would only produce an
+  // error nobody can act on.
+  const canCreateDir = Boolean(currentPath);
+
   const baseLabel = {
     idle: t('status.idle'),
     discovering: t('status.discovering'),
@@ -1875,9 +1894,11 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth,
                 <input type="file" multiple style="display:none"
                   onChange=${uploadFile} />
               </label>
+              ${canCreateDir && html`
               <button class="tb-btn" onClick=${makeDirectory}>
                 <${Icon} name="folder-plus" /> ${t('group.mkdir')}
               </button>
+              `}
             </div>
 
             <div class="breadcrumbs">
@@ -1944,8 +1965,10 @@ function GroupPage({ groupId, group, token, username, userId, onRefreshAuth,
                         onChange=${() => toggle(dirKey(d))} />
                     </td>
                   `}
-                  <td>\u{1F4C1}</td>
-                  <td>${d}/</td>
+                  <td>${unavailableHere.includes(d) ? '\u{26A0}' : '\u{1F4C1}'}</td>
+                  <td>${d}/${unavailableHere.includes(d) ? html`
+                    <span class="root-offline"> ${t('group.root_unavailable')}</span>
+                  ` : ''}</td>
                   <td class="file-size">${inside.length ? formatSize(bytes) : ''}</td>
                   <td class="td-type"></td>
                   <td class="td-date"></td>
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js
index 15401bf..aaed0d1 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/de.js
@@ -92,6 +92,7 @@ export default {
   'group.upload': 'Hochladen',
   'group.mkdir': 'Neuer Ordner',
   'group.mkdir_prompt': 'Name des neuen Ordners:',
+  'group.root_unavailable': '(nicht verfügbar — Laufwerk getrennt)',
   'group.view': 'Ansehen',
   'group.delete': 'Löschen',
   'group.delete_confirm': '{name} löschen?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js
index a98eb1a..02532d6 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/en.js
@@ -92,6 +92,7 @@ export default {
   'group.upload': 'Upload',
   'group.mkdir': 'New folder',
   'group.mkdir_prompt': 'Name of the new folder:',
+  'group.root_unavailable': '(unavailable — the drive is disconnected)',
   'group.view': 'View',
   'group.delete': 'Delete',
   'group.delete_confirm': 'Delete {name}?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js
index 596d952..49bfc1c 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/es.js
@@ -90,6 +90,7 @@ export default {
   'group.upload': 'Subir',
   'group.mkdir': 'Nueva carpeta',
   'group.mkdir_prompt': 'Nombre de la nueva carpeta:',
+  'group.root_unavailable': '(no disponible — la unidad está desconectada)',
   'group.view': 'Ver',
   'group.delete': 'Eliminar',
   'group.delete_confirm': '¿Eliminar {name}?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js
index 61dc99e..1afc5b1 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/fr.js
@@ -91,6 +91,7 @@ export default {
   'group.upload': 'Envoyer',
   'group.mkdir': 'Nouveau dossier',
   'group.mkdir_prompt': 'Nom du nouveau dossier :',
+  'group.root_unavailable': '(indisponible — le disque est déconnecté)',
   'group.view': 'Afficher',
   'group.delete': 'Supprimer',
   'group.delete_confirm': 'Supprimer {name} ?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js
index 6693115..44b4353 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/it.js
@@ -91,6 +91,7 @@ export default {
   'group.upload': 'Carica',
   'group.mkdir': 'Nuova cartella',
   'group.mkdir_prompt': 'Nome della nuova cartella:',
+  'group.root_unavailable': '(non disponibile — l’unità è scollegata)',
   'group.view': 'Visualizza',
   'group.delete': 'Elimina',
   'group.delete_confirm': 'Eliminare {name}?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js
index 8e18a19..33fdee3 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/ja.js
@@ -89,6 +89,7 @@ export default {
   'group.upload': 'アップロード',
   'group.mkdir': '新しいフォルダー',
   'group.mkdir_prompt': '新しいフォルダーの名前：',
+  'group.root_unavailable': '(利用不可 — ドライブが切断されています)',
   'group.view': '表示',
   'group.delete': '削除',
   'group.delete_confirm': '{name} を削除しますか？',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js
index 6fd9220..aa0385e 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/nl.js
@@ -92,6 +92,7 @@ export default {
   'group.upload': 'Uploaden',
   'group.mkdir': 'Nieuwe map',
   'group.mkdir_prompt': 'Naam van de nieuwe map:',
+  'group.root_unavailable': '(niet beschikbaar — de schijf is losgekoppeld)',
   'group.view': 'Bekijken',
   'group.delete': 'Verwijderen',
   'group.delete_confirm': '{name} verwijderen?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js
index 81f0139..e66dd12 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pl.js
@@ -95,6 +95,7 @@ export default {
   'group.upload': 'Wyślij',
   'group.mkdir': 'Nowy folder',
   'group.mkdir_prompt': 'Nazwa nowego folderu:',
+  'group.root_unavailable': '(niedostępne — dysk jest odłączony)',
   'group.view': 'Podgląd',
   'group.delete': 'Usuń',
   'group.delete_confirm': 'Usunąć {name}?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js
index a2f41af..03de040 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/pt-BR.js
@@ -92,6 +92,7 @@ export default {
   'group.upload': 'Enviar',
   'group.mkdir': 'Nova pasta',
   'group.mkdir_prompt': 'Nome da nova pasta:',
+  'group.root_unavailable': '(indisponível — a unidade está desconectada)',
   'group.view': 'Visualizar',
   'group.delete': 'Excluir',
   'group.delete_confirm': 'Excluir {name}?',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js
index 86c7580..72995dd 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/locales/zh-CN.js
@@ -88,6 +88,7 @@ export default {
   'group.upload': '上传',
   'group.mkdir': '新建文件夹',
   'group.mkdir_prompt': '新文件夹的名称：',
+  'group.root_unavailable': '(不可用 — 驱动器已断开连接)',
   'group.view': '查看',
   'group.delete': '删除',
   'group.delete_confirm': '删除 {name}？',
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/style.css b/packages/meshbay-hub/src/meshbay_hub/static/style.css
index 1ba0ef0..6b0b6fe 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/style.css
+++ b/packages/meshbay-hub/src/meshbay_hub/static/style.css
@@ -1894,3 +1894,13 @@ button:disabled { opacity: 0.5; cursor: not-allowed; }
 }
 .btn-primary:hover:not(:disabled) { background: var(--accent-hover); }
 .btn-primary:disabled { opacity: 0.5; cursor: not-allowed; }
+
+/* A root whose volume is absent. Its files stay listed — they are frozen, not
+   deleted — so the row has to carry the difference. */
+.root-offline {
+  font-size: 0.85em;
+  font-weight: 500;
+  color: var(--warn, #d97706);
+  margin-left: 6px;
+}
+.dir-row .root-offline { white-space: nowrap; }
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
diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py
new file mode 100644
index 0000000..03dfb42
--- /dev/null
+++ b/packages/meshbay-node/tests/conftest.py
@@ -0,0 +1,18 @@
+"""Shared fixtures and helpers for node tests."""
+
+from pathlib import Path
+
+from meshbay_node.roots import RootSet
+
+
+def one_root(path: Path, *, name: str = "", kind: str = "generic") -> RootSet:
+    """
+    A RootSet with a single root over `path`, receiving uploads.
+
+    The equivalent of the old `shared_dir`. Note what it implies for assertions:
+    a file directly in `path` now has `entry.path == <basename of path>`, not
+    `""` — every index path carries its root name, in a group with one root as
+    much as in a group with five.
+    """
+    return RootSet.build([{"path": str(path), "name": name, "kind": kind,
+                           "upload": True}])
diff --git a/packages/meshbay-node/tests/test_admin_ops_mnp.py b/packages/meshbay-node/tests/test_admin_ops_mnp.py
new file mode 100644
index 0000000..1bf8365
--- /dev/null
+++ b/packages/meshbay-node/tests/test_admin_ops_mnp.py
@@ -0,0 +1,290 @@
+"""
+`gek_rotate` and `member_unpin` over MNP.
+
+Both are destructive and both are new, so the tests are negative assertions:
+nobody without the operator's pinned key can reach them, a signature over the
+wrong transcript does not count, and the operation cannot be triggered by the
+request message alone.
+
+The rule these live under is worth restating, because it is easy to read
+draft-v5 §5.1 as forbidding them: **"nothing arriving over MNP can activate a
+GEK" is about key material arriving from outside** (C5b — a member handing the
+node a key of their choosing). An operator-signed instruction where the node
+generates the key with its own CSPRNG is a different shape, and it is the only
+thing that finishes a revocation: the ex-member still holds the current key.
+"""
+
+import base64
+import time
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from conftest import one_root
+from meshbay_common.adminop import (
+    OP_GEK_ROTATE,
+    OP_MEMBER_UNPIN,
+    admin_transcript,
+)
+from meshbay_common.crypto import pk_to_b64
+from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import open_roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+GROUP = "g" * 32
+
+
+@pytest.fixture
+async def roster(tmp_path):
+    r = await open_roster(tmp_path)
+    yield r
+    await r.close()
+
+
+def _keypair():
+    sk = Ed25519PrivateKey.generate()
+    return sk, pk_to_b64(sk.public_key())
+
+
+async def _session(tmp_path: Path, roster, *, operator: bool) -> WebRTCPeerSession:
+    """A peer session with an operator pinned, or deliberately without one."""
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roots = one_root(shared)
+
+    sk_op, pk_op = _keypair()
+    if operator:
+        await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
+        await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")
+
+    group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
+                 "join_policy": "invite"}
+    state = {
+        "groups_ctx": {GROUP: group_ctx},
+        "roster": roster,
+        "indexes": {GROUP: index},
+        "bundle_store": _FakeBundleStore(),
+        "pk_x25519_raw": b"\x02" * 32,
+        "hub": _FakeHub(),
+        "node_user_id": "node-user",
+    }
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {
+        "roots": roots, "index": index, "sk_node": index.sk_node,
+        "roster": roster, "groups": {GROUP: group_ctx},
+        "has_admin_authority": operator,
+        "daemon_state": state,
+    }
+    session._group_id = GROUP
+    session._user_id = "grenet" if operator else "mallory"
+    session._username = session._user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session._admin_ops = {}
+    session._remote_ip = ""
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    session.state = state
+    session.sk_op = sk_op
+    session.spawned = []
+    session._spawn = session.spawned.append
+    return session
+
+
+class _FakeBundleStore:
+    def __init__(self):
+        self.stored = []
+
+    async def store(self, *args):
+        self.stored.append(args)
+
+
+class _FakeHub:
+    class _S:
+        user_id = "node-user"
+    _session = _S()
+
+
+def _last(session):
+    return session.sent[-1] if session.sent else {}
+
+
+async def _drain(session):
+    """Await whatever `_spawn` started. The real session holds its tasks; a
+    hand-built one collects them here so the assertion sees the result."""
+    for coro in session.spawned:
+        await coro
+    session.spawned.clear()
+
+
+async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn):
+    challenge = _last(session)
+    assert challenge["type"] == "admin_challenge", challenge
+    transcript = admin_transcript(
+        op=op, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
+        subject=subject, nonce=base64.b64decode(challenge["nonce"]),
+        ts=challenge["ts"])
+    pending = session._admin_ops.get(challenge["op_id"]) or {
+        "op": op, "subject": subject}
+    await exec_fn(pending, transcript, sk.sign(transcript))
+
+
+# ── gek_rotate ───────────────────────────────────────────────────────────────
+
+async def test_rotation_needs_an_operator(tmp_path, roster):
+    """Without a paired operator there is nobody who could sign, so the node
+    fails closed and says why rather than issuing a challenge nobody can meet."""
+    session = await _session(tmp_path, roster, operator=False)
+
+    session._do_gek_rotate({})
+
+    assert _last(session)["type"] == "error"
+    assert "authorized key" in _last(session)["detail"]
+    assert not session._admin_ops
+
+
+async def test_the_request_alone_rotates_nothing(tmp_path, roster):
+    """The message asks; only a signature acts. A node that rotated here would
+    let any member lock the group out."""
+    session = await _session(tmp_path, roster, operator=True)
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+
+    assert _last(session)["type"] == "admin_challenge"
+    assert session._ctx["groups"][GROUP]["gek"] == before
+
+
+async def test_a_members_signature_does_not_rotate(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    sk_mallory, pk_mallory = _keypair()
+    await roster.pin_identity("mallory", "mallory", pk_mallory, pk_mallory, "code")
+    await roster.set_member(GROUP, "mallory", ROLE_MEMBER, "active", "grenet")
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, sk_mallory,
+                         session._admin_exec_gek_rotate)
+
+    assert _last(session)["type"] == "error"
+    assert session._ctx["groups"][GROUP]["gek"] == before
+
+
+async def test_a_signature_over_another_operation_does_not_count(tmp_path, roster):
+    """
+    H5's rule: the node rebuilds the transcript from the operation it is holding
+    and verifies against *that*, so a signature collected for one act cannot be
+    presented as another.
+
+    Driven through `_do_admin_response`, deliberately. Handing a transcript
+    straight to `_admin_exec_*` would skip the reconstruction that is the
+    control, and the test would pass while proving nothing.
+    """
+    session = await _session(tmp_path, roster, operator=True)
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+    challenge = _last(session)
+
+    # Signed over member_unpin, presented against the pending gek_rotate.
+    wrong = admin_transcript(
+        op=OP_MEMBER_UNPIN, node_pk_b64=session._node_pk_b64(), group_id=GROUP,
+        subject=GROUP, nonce=base64.b64decode(challenge["nonce"]),
+        ts=challenge["ts"])
+    session._do_admin_response({
+        "op_id": challenge["op_id"],
+        "signature": base64.b64encode(session.sk_op.sign(wrong)).decode(),
+    })
+    await _drain(session)
+
+    assert _last(session)["type"] == "error"
+    assert session.state["groups_ctx"][GROUP]["gek"] == before
+
+
+async def test_the_operator_rotates_and_the_node_makes_the_key(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    before = session._ctx["groups"][GROUP]["gek"]
+
+    session._do_gek_rotate({})
+    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op,
+                         session._admin_exec_gek_rotate)
+
+    ack = _last(session)
+    assert ack["type"] == MNP.GEK_ROTATE_ACK, ack
+    after = session.state["groups_ctx"][GROUP]["gek"]
+    assert after != before, "the key did not change"
+    assert len(after) == 32
+    # Produced here, not received: no key material crossed the wire (C5b).
+    assert session.state["bundle_store"].stored, (
+        "the node's own copy was not stored — the daemon could not reload it")
+
+
+async def test_rotation_reaches_the_index(tmp_path, roster):
+    """The index is encrypted under the GEK. Leaving the old key on it would
+    serve members a listing they cannot open."""
+    session = await _session(tmp_path, roster, operator=True)
+
+    session._do_gek_rotate({})
+    await _sign_and_exec(session, OP_GEK_ROTATE, GROUP, session.sk_op,
+                         session._admin_exec_gek_rotate)
+
+    assert session.state["indexes"][GROUP].gek == \
+        session.state["groups_ctx"][GROUP]["gek"]
+
+
+# ── member_unpin ─────────────────────────────────────────────────────────────
+
+async def test_unpinning_needs_an_operator(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=False)
+    session._do_member_unpin({"user_id": "bob"})
+    assert _last(session)["type"] == "error"
+
+
+async def test_unpinning_yourself_is_refused(tmp_path, roster):
+    """It would end the authority of the connection performing the operation,
+    halfway through it."""
+    session = await _session(tmp_path, roster, operator=True)
+    session._do_member_unpin({"user_id": "grenet"})
+    assert _last(session)["detail"] == "Cannot unpin yourself"
+
+
+async def test_a_members_signature_does_not_unpin(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    sk_bob, pk_bob = _keypair()
+    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
+
+    session._do_member_unpin({"user_id": "bob"})
+    await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", sk_bob,
+                         session._admin_exec_member_unpin)
+
+    assert _last(session)["type"] == "error"
+    assert await roster.get_identity("bob") is not None, (
+        "a member removed their own pin — only the operator may")
+
+
+async def test_the_operator_unpins(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    _, pk_bob = _keypair()
+    await roster.pin_identity("bob", "bob", pk_bob, pk_bob, "code")
+
+    session._do_member_unpin({"user_id": "bob"})
+    await _sign_and_exec(session, OP_MEMBER_UNPIN, "bob", session.sk_op,
+                         session._admin_exec_member_unpin)
+
+    assert _last(session)["type"] == MNP.MEMBER_UNPIN_ACK
+    assert await roster.get_identity("bob") is None
+
+
+async def test_unpinning_someone_unknown_says_so(tmp_path, roster):
+    session = await _session(tmp_path, roster, operator=True)
+    session._do_member_unpin({"user_id": "nobody"})
+    await _sign_and_exec(session, OP_MEMBER_UNPIN, "nobody", session.sk_op,
+                         session._admin_exec_member_unpin)
+    assert _last(session)["type"] == "error"
+    assert "No such pinned identity" in _last(session)["detail"]
diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
new file mode 100644
index 0000000..faca0ce
--- /dev/null
+++ b/packages/meshbay-node/tests/test_cli_dispatch.py
@@ -0,0 +1,129 @@
+"""
+Every CLI verb reaches its own code without falling over on a name.
+
+`meshbay-node reload` shipped with `subprocess` unimported and crashed with a
+NameError the first time it was typed. Python compiles that file happily —
+`node --check validates syntax, not names` is already in CLAUDE.md about the
+SPA, and it is the same class here: nothing in the module is wrong until the
+branch runs.
+
+So this walks every documented verb with the daemon stubbed out, and asserts
+that the branch executes. It is deliberately shallow — what each command *does*
+is tested in `test_ops.py` and `test_roster_pairing.py`. What this catches is a
+branch nobody ever ran.
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+from meshbay_node import daemon as daemon_mod
+
+# Each verb, with the arguments that reach its branch. `--yes` where the command
+# would otherwise stop for a confirmation nobody can type in a test.
+VERBS = [
+    ["status"],
+    ["ui"],
+    ["group", "list"],
+    ["group", "add"],                      # missing --dir: usage, then exit
+    ["gek", "init"],
+    ["gek", "rotate", "--yes"],
+    ["gek-init"],
+    ["member", "list"],
+    ["member", "invite", "bob"],
+    ["member", "revoke", "bob"],
+    ["member", "unpin", "bob"],
+    ["operator", "pair"],
+    ["file", "list"],
+    ["file", "rm", "abc", "--yes"],
+    ["denylist", "show"],
+    ["denylist", "clear", "--yes"],
+    ["reload"],
+]
+
+
+@pytest.fixture
+def stub_daemon(monkeypatch, tmp_path):
+    """
+    Answer every loopback call with an empty-ish payload.
+
+    The point is to reach the branch, not to exercise the daemon: a command that
+    only crashes when the node is running is still a command that crashes.
+    """
+    calls: list[tuple] = []
+
+    def fake_api(cfg, path, method="GET", timeout=30, body=None):
+        calls.append((method, path))
+        return {
+            "groups": [], "files": [], "identities": [], "members": [],
+            "invites": [], "users": [], "jtis": [], "count": 0, "removed": 0,
+            "subject": "all", "code": "TEST-CODE", "expires_at": "",
+            "user_id": "u", "authorized_members": 0, "errors": [],
+            "name": "g", "group_id": "g", "shared_dir": str(tmp_path),
+            "config": str(tmp_path / "node.toml"),
+        }
+
+    monkeypatch.setattr(daemon_mod, "_daemon_api", fake_api)
+    monkeypatch.setattr(daemon_mod, "_resolve_group", lambda cfg, g: "g" * 32)
+
+    conf = tmp_path / "node.toml"
+    conf.write_text('[hub]\nurl = "https://example.invalid"\n')
+    monkeypatch.setattr(daemon_mod, "DEFAULT_CONFIG_PATH", conf)
+
+    # No interactive prompt left to hang on.
+    monkeypatch.setattr("builtins.input", lambda *a: "n")
+
+    # `reload` looks for a real daemon and signals it. Without this the test
+    # SIGHUPs whatever node happens to be running on the machine — which it did,
+    # once, before this was added. A test must not reach outside itself.
+    import os
+    import subprocess
+
+    signalled: list[int] = []
+    monkeypatch.setattr(
+        subprocess, "run",
+        lambda *a, **k: subprocess.CompletedProcess(a[0], 0, stdout="4242\n",
+                                                    stderr=""))
+    monkeypatch.setattr(os, "kill", lambda pid, sig: signalled.append(pid))
+    calls.append(("_signalled", signalled))
+    return calls
+
+
+@pytest.mark.parametrize("argv", VERBS, ids=lambda a: "-".join(a))
+def test_every_verb_reaches_its_branch(argv, stub_daemon, monkeypatch, capsys):
+    monkeypatch.setattr(sys, "argv", ["meshbay-node", *argv])
+    try:
+        daemon_mod.main()
+    except SystemExit:
+        # A usage message and exit(1) is a branch that ran, which is what this
+        # asserts. A NameError or AttributeError is not.
+        pass
+    except (NameError, AttributeError) as e:            # pragma: no cover
+        pytest.fail(f"{' '.join(argv)} crashed on a name: {e}")
+
+    out = capsys.readouterr()
+    assert out.out or out.err, f"{' '.join(argv)} printed nothing at all"
+
+
+def test_the_verb_list_here_matches_the_parser():
+    """
+    A verb added to the parser and not to this file would go untested, which is
+    exactly how `reload` shipped broken.
+    """
+    import argparse
+    import inspect
+
+    source = inspect.getsource(daemon_mod.main)
+    start = source.index('choices=[') + len('choices=[')
+    end = source.index(']', start)
+    declared = {c.strip().strip('"\'') for c in source[start:end].split(',')
+                if c.strip()}
+
+    exercised = {argv[0] for argv in VERBS}
+    # `init` writes a config file and `calibrate-argon2` burns CPU for seconds;
+    # both are excluded on purpose rather than by omission.
+    untested = declared - exercised - {"init", "calibrate-argon2"}
+    assert not untested, (
+        f"CLI verbs with no dispatch test: {sorted(untested)} — add them to "
+        f"VERBS above")
diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index 60ef11f..7974be5 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -18,6 +18,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 from meshbay_common.crypto import generate_gek
 from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
+from conftest import one_root
 from meshbay_node.daemon import NodeDaemon
 from meshbay_node.indexer import DirectoryIndexer
 
@@ -222,7 +223,7 @@ async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hu
 
     sk_node = Ed25519PrivateKey.generate()
     indexer = DirectoryIndexer(
-        root=shared_dir, group_id="a" * 32,
+        roots=one_root(shared_dir), group_id="a" * 32,
         sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
@@ -272,7 +273,7 @@ async def test_daemon_index_change_registers_swarm_for_public_group(
     daemon._state["endpoint_hint"] = "node123"
 
     indexer = DirectoryIndexer(
-        root=shared_dir, group_id="a" * 32,
+        roots=one_root(shared_dir), group_id="a" * 32,
         sk_node=Ed25519PrivateKey.generate(), gek=gek)
     await indexer.initial_scan()
 
@@ -302,7 +303,7 @@ async def test_daemon_index_change_skips_other_group_peers(
 
     sk_node = Ed25519PrivateKey.generate()
     indexer = DirectoryIndexer(
-        root=shared_dir, group_id="a" * 32,
+        roots=one_root(shared_dir), group_id="a" * 32,
         sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py
index 380d752..c304361 100644
--- a/packages/meshbay-node/tests/test_indexer.py
+++ b/packages/meshbay-node/tests/test_indexer.py
@@ -10,6 +10,7 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 
 from meshbay_common.crypto import generate_gek
 from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from conftest import one_root
 from meshbay_node.keystore import NodeKeys
 
 
@@ -115,7 +116,7 @@ def test_group_index_diff(sk_node, gek):
 @pytest.mark.asyncio
 async def test_initial_scan(shared_dir, sk_node, gek):
     indexer = DirectoryIndexer(
-        root=shared_dir,
+        roots=one_root(shared_dir),
         group_id="scan-test",
         sk_node=sk_node,
         gek=gek,
@@ -134,7 +135,7 @@ async def test_initial_scan(shared_dir, sk_node, gek):
 
 @pytest.mark.asyncio
 async def test_type_detection(shared_dir, sk_node, gek):
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     by_name = {e.name: e.type for e in indexer.index.entries}
@@ -152,7 +153,7 @@ async def test_hidden_files_excluded(tmp_path, sk_node, gek):
     (d / "visible.txt").write_bytes(b"visible")
     (d / "file.tmp").write_bytes(b"tmp")
 
-    indexer = DirectoryIndexer(root=d, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     names = {e.name for e in indexer.index.entries}
@@ -169,7 +170,7 @@ async def test_on_change_callback(shared_dir, sk_node, gek):
         changes.append(idx.index.count)
 
     indexer = DirectoryIndexer(
-        root=shared_dir, group_id="g", sk_node=sk_node, gek=gek,
+        roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek,
         on_change=on_change)
     await indexer.start()
     await asyncio.sleep(0.1)
@@ -183,7 +184,7 @@ async def test_on_change_callback(shared_dir, sk_node, gek):
 
 @pytest.mark.asyncio
 async def test_index_roundtrip_after_scan(shared_dir, sk_node, gek):
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     wire = indexer.index.serialize()
diff --git a/packages/meshbay-node/tests/test_multi_group.py b/packages/meshbay-node/tests/test_multi_group.py
index 9be8d47..2828d08 100644
--- a/packages/meshbay-node/tests/test_multi_group.py
+++ b/packages/meshbay-node/tests/test_multi_group.py
@@ -17,6 +17,7 @@ from cryptography.hazmat.primitives import serialization
 
 from meshbay_common.crypto import generate_gek, pk_to_b64
 from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from conftest import one_root
 from meshbay_node.transport.quic_server import QuicChunkServer
 from meshbay_node.transport.quic_client import QuicChunkClient
 
@@ -72,15 +73,15 @@ async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_pa
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer_a = DirectoryIndexer(root=dir_a, group_id="group-a", sk_node=sk_node, gek=gek_a)
+    indexer_a = DirectoryIndexer(roots=one_root(dir_a), group_id="group-a", sk_node=sk_node, gek=gek_a)
     await indexer_a.initial_scan()
 
-    indexer_b = DirectoryIndexer(root=dir_b, group_id="group-b", sk_node=sk_node, gek=gek_b)
+    indexer_b = DirectoryIndexer(roots=one_root(dir_b), group_id="group-b", sk_node=sk_node, gek=gek_b)
     await indexer_b.initial_scan()
 
     groups = {
-        "group-a": {"gek": gek_a, "shared_root": dir_a, "index": indexer_a.index},
-        "group-b": {"gek": gek_b, "shared_root": dir_b, "index": indexer_b.index},
+        "group-a": {"gek": gek_a, "roots": dir_a, "index": indexer_a.index},
+        "group-b": {"gek": gek_b, "roots": dir_b, "index": indexer_b.index},
     }
 
     cert_path = tmp_path / "node.crt"
@@ -88,7 +89,7 @@ async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_pa
 
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem,
-        gek=gek_a, shared_root=dir_a, index=indexer_a.index,
+        gek=gek_a, roots=one_root(dir_a), index=indexer_a.index,
         host="127.0.0.1", port=19200,
         cert_path=cert_path, key_path=key_path,
         groups=groups,
diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py
new file mode 100644
index 0000000..f7fd259
--- /dev/null
+++ b/packages/meshbay-node/tests/test_ops.py
@@ -0,0 +1,179 @@
+"""
+One implementation, several front doors.
+
+The point of `meshbay_node.ops` is not tidiness. C1 and C6 were both "a second
+path into the node with its own weaker handshake", and two implementations of
+`revoke` with two authorization checks is that shape one size down. So the tests
+that matter here are the ones that would fail if a second implementation
+appeared: the adapters must be thin, and the operations must not decide who may
+call them.
+"""
+
+import inspect
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from conftest import one_root
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.transport.quic_server import Denylist
+
+
+
+def _state(tmp_path: Path) -> dict:
+    shared = tmp_path / "shared"
+    shared.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    return {
+        "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(shared),
+                                  "gek": None}},
+        "denylist": Denylist(path=tmp_path / "denylist.json"),
+        "indexes": {"g" * 32: index},
+    }
+
+
+# ── The adapters stay thin ───────────────────────────────────────────────────
+
+def test_operations_take_state_and_nothing_web_shaped():
+    """
+    An operation that knew about HTTP could not be called from MNP without a
+    second copy. Every public operation therefore takes `state` first and
+    returns plain data.
+    """
+    public = [(n, f) for n, f in vars(ops).items()
+              if inspect.iscoroutinefunction(f) and not n.startswith("_")]
+    assert public, "no operations found — did the module move?"
+    for name, fn in public:
+        params = list(inspect.signature(fn).parameters)
+        assert params and params[0] == "state", (
+            f"{name} does not take state first — an adapter would have to "
+            f"assemble something for it, which is where a second implementation "
+            f"begins")
+
+
+def test_the_http_adapter_adds_no_logic():
+    """
+    Each loopback handler should be a call into `ops` and nothing else. A
+    handler that grew a check of its own would be a rule the MNP path does not
+    have.
+    """
+    import meshbay_node.ui.app as ui
+    source = inspect.getsource(ui)
+    # Every endpoint that performs an operation routes through _op(...).
+    for endpoint in ("operator_pair", "create_invite", "revoke_member",
+                     "unpin_member", "init_gek", "attach_group", "delete_file"):
+        start = source.index(f"async def {endpoint}(")
+        body = source[start:start + 700]
+        assert "_op(" in body.split("\n\n")[0] + body, (
+            f"{endpoint} does not go through the shared adapter")
+        assert "roster.set_status" not in body and "generate_gek" not in body, (
+            f"{endpoint} performs the operation itself instead of calling ops")
+
+
+def test_op_errors_carry_a_status_without_importing_http():
+    source = inspect.getsource(ops)
+    for forbidden in ("JSONResponse", "fastapi", "starlette", "HTTPException"):
+        assert forbidden not in source, (
+            f"ops imports {forbidden} — it must not know which adapter called it")
+
+
+# ── Denylist (14.10) ─────────────────────────────────────────────────────────
+
+async def test_denylist_reports_what_it_refuses(tmp_path):
+    state = _state(tmp_path)
+    state["denylist"].deny_user("alice")
+    state["denylist"].deny_group("g" * 32)
+
+    out = await ops.read_denylist(state)
+
+    assert out["users"] == ["alice"]
+    assert out["groups"] == ["g" * 32]
+    assert out["count"] == 2
+
+
+async def test_clearing_one_subject_leaves_the_rest(tmp_path):
+    state = _state(tmp_path)
+    state["denylist"].deny_user("alice")
+    state["denylist"].deny_user("bob")
+
+    out = await ops.clear_denylist(state, subject="alice")
+
+    assert out["removed"] == 1
+    assert (await ops.read_denylist(state))["users"] == ["bob"]
+
+
+async def test_clearing_everything_says_how_much(tmp_path):
+    """The count is the point: it tells the operator whether they undid one
+    revocation or all of them."""
+    state = _state(tmp_path)
+    for name in ("a", "b", "c"):
+        state["denylist"].deny_user(name)
+
+    out = await ops.clear_denylist(state)
+
+    assert out["removed"] == 3
+    assert (await ops.read_denylist(state))["count"] == 0
+
+
+async def test_denylist_survives_a_restart(tmp_path):
+    """Finding H4: revocations used to live only in memory, so a restart
+    silently un-revoked everyone."""
+    state = _state(tmp_path)
+    state["denylist"].deny_user("alice")
+
+    reopened = Denylist(path=tmp_path / "denylist.json")
+    assert reopened.is_denied("alice", jti="", group_id="")
+
+
+# ── File deletion (14.11) ────────────────────────────────────────────────────
+
+async def test_deleting_a_file_removes_it_from_disk_and_index(tmp_path):
+    state = _state(tmp_path)
+    ctx = state["groups_ctx"]["g" * 32]
+    target = ctx["roots"].roots[0].path / "gone.txt"
+    target.write_text("x")
+    from meshbay_common.protocol import IndexEntry
+    ctx["index"].add_entry(IndexEntry(id="a" * 64, name="gone.txt", path="shared",
+                                      size=1, type="other", added_at=0))
+
+    out = await ops.delete_file(state, "g" * 32, "a" * 64)
+
+    assert out["status"] == "deleted"
+    assert not target.exists()
+    assert ctx["index"].get_entry("a" * 64) is None
+
+
+async def test_deleting_from_an_unavailable_root_is_refused(tmp_path):
+    """
+    A frozen root's files are still listed. Deleting one would either fail
+    obscurely or — worse, once the drive returns — leave the index and the disk
+    disagreeing.
+    """
+    state = _state(tmp_path)
+    ctx = state["groups_ctx"]["g" * 32]
+    from meshbay_common.protocol import IndexEntry
+    ctx["index"].add_entry(IndexEntry(id="a" * 64, name="frozen.txt", path="shared",
+                                      size=1, type="other", added_at=0))
+    ctx["roots"].roots[0].available = False
+
+    with pytest.raises(ops.OpError, match="frozen, not gone"):
+        await ops.delete_file(state, "g" * 32, "a" * 64)
+
+    assert ctx["index"].get_entry("a" * 64) is not None
+
+
+async def test_deleting_an_unknown_file_says_so(tmp_path):
+    state = _state(tmp_path)
+    with pytest.raises(ops.OpError, match="No such file"):
+        await ops.delete_file(state, "g" * 32, "f" * 64)
+
+
+async def test_an_unhosted_group_offers_what_it_does_host(tmp_path):
+    """A bare "no such group" leaves an operator guessing at a UUID."""
+    state = _state(tmp_path)
+    with pytest.raises(ops.OpError) as exc:
+        await ops.delete_file(state, "z" * 32, "a" * 64)
+    assert exc.value.status == 404
+    assert exc.value.extra.get("available")
diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py
index 93ab1b0..5720043 100644
--- a/packages/meshbay-node/tests/test_quic_transport.py
+++ b/packages/meshbay-node/tests/test_quic_transport.py
@@ -14,6 +14,7 @@ from cryptography.hazmat.primitives import serialization
 
 from meshbay_common.crypto import generate_gek, pk_to_b64
 from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from conftest import one_root
 from meshbay_node.transport.quic_server import QuicChunkServer, Denylist
 from meshbay_node.transport.quic_client import QuicChunkClient
 
@@ -60,7 +61,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path):
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     cert_path = tmp_path / "node.crt"
@@ -68,7 +69,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path):
 
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         host="127.0.0.1", port=19100,
         cert_path=cert_path, key_path=key_path,
     )
@@ -98,7 +99,7 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     cert_path = tmp_path / "node.crt"
@@ -106,7 +107,7 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
 
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         host="127.0.0.1", port=19101,
         cert_path=cert_path, key_path=key_path,
     )
@@ -133,7 +134,7 @@ async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     cert_path = tmp_path / "node.crt"
@@ -141,7 +142,7 @@ async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p
 
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         host="127.0.0.1", port=19102,
         cert_path=cert_path, key_path=key_path,
     )
@@ -167,7 +168,7 @@ async def test_quic_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     cert_path = tmp_path / "node.crt"
@@ -175,7 +176,7 @@ async def test_quic_wrong_group_rejected(sk_node, sk_hub, gek, shared_dir, tmp_p
 
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         host="127.0.0.1", port=19103,
         cert_path=cert_path, key_path=key_path,
     )
@@ -201,7 +202,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     cert_path = tmp_path / "node.crt"
@@ -209,7 +210,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat
 
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         host="127.0.0.1", port=19104,
         cert_path=cert_path, key_path=key_path,
     )
@@ -255,7 +256,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p
     hub_pk_pem = sk_hub.public_key().public_bytes(
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     cert_path = tmp_path / "node.crt"
@@ -264,7 +265,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p
     denylist = Denylist()
     server = QuicChunkServer(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         host="127.0.0.1", port=19105,
         cert_path=cert_path, key_path=key_path,
         denylist=denylist,
diff --git a/packages/meshbay-node/tests/test_root_availability.py b/packages/meshbay-node/tests/test_root_availability.py
new file mode 100644
index 0000000..028c308
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_availability.py
@@ -0,0 +1,301 @@
+"""
+A root that goes away freezes; it never empties.
+
+This is the property the whole per-root availability design exists for. Unplug a
+drive while the node is running and the filesystem watcher either reports every
+file under it as deleted, or the next scan sees an empty directory. Acting on
+either propagates deletions for a whole library, to every member, as though the
+owner had erased it — and the index is what the node serves, so the loss is not
+local.
+
+Every test here is written as "the entries are still there". They fail against
+an indexer that treats a vanished root as a set of deletions, which is what the
+straightforward implementation does.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+
+pytestmark = pytest.mark.asyncio
+
+
+def _roots(*paths: Path) -> RootSet:
+    specs = [{"path": str(p)} for p in paths]
+    specs[0]["upload"] = True
+    return RootSet.build(specs)
+
+
+async def _indexer(roots: RootSet) -> DirectoryIndexer:
+    idx = DirectoryIndexer(roots=roots, group_id="g" * 32,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    return idx
+
+
+def _names(idx: DirectoryIndexer) -> set[str]:
+    return {e.name for e in idx.index.entries}
+
+
+# ── The freeze ───────────────────────────────────────────────────────────────
+
+async def test_a_vanished_root_does_not_empty_the_index(tmp_path):
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    (films / "b.mkv").write_bytes(b"b")
+
+    idx = await _indexer(_roots(films))
+    assert _names(idx) == {"a.mkv", "b.mkv"}
+
+    # The volume goes away. Watchdog would now report both files as deleted.
+    for f in films.iterdir():
+        f.unlink()
+    films.rmdir()
+
+    for f in ("a.mkv", "b.mkv"):
+        await idx._update_entry(films / f, deleted=True)
+
+    assert _names(idx) == {"a.mkv", "b.mkv"}, (
+        "an unplugged drive emptied the index — every member would see the "
+        "library as deleted")
+    assert idx.roots.roots[0].available is False
+
+
+async def test_reconcile_does_not_delete_from_an_unavailable_root(tmp_path):
+    """The sweep must skip roots it cannot read: there is nothing to compare
+    against, and comparing anyway deletes everything."""
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    idx = await _indexer(_roots(films))
+    (films / "a.mkv").unlink()
+    films.rmdir()
+
+    await idx.reconcile()
+
+    assert _names(idx) == {"a.mkv"}
+    assert idx.roots.roots[0].available is False
+
+
+async def test_one_root_going_away_leaves_the_others_alone(tmp_path):
+    films = tmp_path / "Films"
+    music = tmp_path / "Music"
+    films.mkdir()
+    music.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    (music / "b.mp3").write_bytes(b"b")
+
+    idx = await _indexer(_roots(films, music))
+    assert _names(idx) == {"a.mkv", "b.mp3"}
+
+    (music / "b.mp3").unlink()
+    music.rmdir()
+    await idx.reconcile()
+
+    assert _names(idx) == {"a.mkv", "b.mp3"}
+    by_name = {r.name: r.available for r in idx.roots}
+    assert by_name == {"Films": True, "Music": False}
+
+
+async def test_members_are_told_which_roots_are_unavailable(tmp_path):
+    """Frozen entries stay listed, so without this a member cannot tell
+    "temporarily unavailable" from "still there"."""
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    idx = await _indexer(_roots(films))
+    assert idx.index.roots == [
+        {"name": "Films", "kind": "generic", "available": True, "upload": True}]
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+
+    assert idx.index.roots[0]["available"] is False
+
+
+# ── The counter-property ─────────────────────────────────────────────────────
+
+async def test_a_file_deleted_from_a_live_root_is_removed(tmp_path):
+    """
+    The freeze must not become "deletions never happen". A root that is readable
+    and a file that is genuinely gone is an ordinary deletion.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    (films / "b.mkv").write_bytes(b"b")
+
+    idx = await _indexer(_roots(films))
+    (films / "a.mkv").unlink()
+    await idx._update_entry(films / "a.mkv", deleted=True)
+
+    assert _names(idx) == {"b.mkv"}
+
+
+async def test_reconcile_removes_what_is_genuinely_gone(tmp_path):
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    (films / "b.mkv").write_bytes(b"b")
+
+    idx = await _indexer(_roots(films))
+    (films / "a.mkv").unlink()
+    await idx.reconcile()
+
+    assert _names(idx) == {"b.mkv"}
+
+
+async def test_reconcile_picks_up_a_file_the_watcher_missed(tmp_path):
+    """
+    `ReadDirectoryChangesW` drops events under load and inotify on a FUSE mount
+    misses changes made outside it. Both are the common case here, so the sweep
+    is the only thing that recovers.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    idx = await _indexer(_roots(films))
+
+    (films / "late.mkv").write_bytes(b"x")     # no event delivered
+    await idx.reconcile()
+
+    assert _names(idx) == {"late.mkv"}
+
+
+async def test_a_returning_root_is_rescanned(tmp_path):
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    idx = await _indexer(_roots(films))
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+    assert _names(idx) == {"a.mkv"}          # frozen
+
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    (films / "c.mkv").write_bytes(b"c")
+    await idx.reconcile()
+
+    assert _names(idx) == {"a.mkv", "c.mkv"}
+    assert idx.roots.roots[0].available is True
+
+
+async def test_a_root_absent_at_startup_is_not_an_error(tmp_path):
+    """
+    Someone starts the node with the drive unplugged. The group still exists and
+    the other roots still serve; this one fills in when it returns.
+    """
+    films = tmp_path / "Films"
+    music = tmp_path / "Music"
+    films.mkdir()
+    music.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    roots = _roots(films, music)
+    music.rmdir()
+
+    idx = await _indexer(roots)
+
+    assert _names(idx) == {"a.mkv"}
+    assert {r.name: r.available for r in idx.roots} == {"Films": True, "Music": False}
+
+
+# ── Paths carry their root ───────────────────────────────────────────────────
+
+async def test_every_path_starts_with_its_root_name(tmp_path):
+    films = tmp_path / "Films"
+    (films / "2024").mkdir(parents=True)
+    (films / "top.mkv").write_bytes(b"t")
+    (films / "2024" / "deep.mkv").write_bytes(b"d")
+
+    idx = await _indexer(_roots(films))
+    by_name = {e.name: e.path for e in idx.index.entries}
+
+    assert by_name == {"top.mkv": "Films", "deep.mkv": "Films/2024"}
+
+
+async def test_a_single_root_group_is_not_a_special_case(tmp_path):
+    """One path shape has to be got right once; two have to be kept right
+    forever. A lone root prefixes exactly like any other."""
+    only = tmp_path / "Shared"
+    only.mkdir()
+    (only / "x.txt").write_bytes(b"x")
+
+    idx = await _indexer(_roots(only))
+    assert [e.path for e in idx.index.entries] == ["Shared"]
+
+
+async def test_same_relative_path_in_two_roots_stays_distinct(tmp_path):
+    films = tmp_path / "Films"
+    music = tmp_path / "Music"
+    (films / "2024").mkdir(parents=True)
+    (music / "2024").mkdir(parents=True)
+    (films / "2024" / "same.dat").write_bytes(b"film")
+    (music / "2024" / "same.dat").write_bytes(b"music")
+
+    idx = await _indexer(_roots(films, music))
+    paths = sorted(e.path for e in idx.index.entries)
+
+    assert paths == ["Films/2024", "Music/2024"]
+    assert len(idx.index.entries) == 2
+
+
+# ── Duplicate content ────────────────────────────────────────────────────────
+
+async def test_identical_files_do_not_churn_the_index(tmp_path):
+    """
+    The index is keyed by content hash, so the same bytes at two paths are one
+    entry. Reconciliation compares paths, so without care it decides the second
+    path is a missed event **every cycle** — rewriting that entry, bumping the
+    version, and pushing an index update to every connected peer once a minute.
+
+    Found on a live node: `clip.mp4` sat at the root of a shared directory and
+    in `uploads/` with identical bytes.
+    """
+    films = tmp_path / "Films"
+    (films / "uploads").mkdir(parents=True)
+    (films / "clip.mp4").write_bytes(b"same bytes")
+    (films / "uploads" / "clip.mp4").write_bytes(b"same bytes")
+
+    idx = await _indexer(_roots(films))
+    assert len(idx.index.entries) == 1, "content-addressed index, so one entry"
+
+    await idx.reconcile()
+    first = (idx.index.version, idx.index.entries[0].path)
+    await idx.reconcile()
+    second = (idx.index.version, idx.index.entries[0].path)
+
+    assert first == second, (
+        "reconciliation rewrote the entry for a duplicate it cannot represent — "
+        "every peer would receive an index update every cycle")
+
+
+async def test_deleting_one_copy_keeps_the_other_listed(tmp_path):
+    """
+    The mirror case: the recorded path goes, identical content stays. Dropping
+    the entry would delist a file that is still on disk and still servable.
+    """
+    films = tmp_path / "Films"
+    (films / "uploads").mkdir(parents=True)
+    (films / "clip.mp4").write_bytes(b"same bytes")
+    (films / "uploads" / "clip.mp4").write_bytes(b"same bytes")
+
+    idx = await _indexer(_roots(films))
+    recorded = idx.index.entries[0].path
+    survivor = "Films/uploads" if recorded == "Films" else "Films"
+
+    (films / "clip.mp4").unlink() if recorded == "Films" else \
+        (films / "uploads" / "clip.mp4").unlink()
+    await idx.reconcile()
+
+    assert len(idx.index.entries) == 1, "the surviving copy was delisted"
+    assert idx.index.entries[0].path == survivor
diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py
new file mode 100644
index 0000000..ea4ba6a
--- /dev/null
+++ b/packages/meshbay-node/tests/test_roots.py
@@ -0,0 +1,242 @@
+"""
+Several named roots per group.
+
+Most of these are negative assertions — a root set that would be ambiguous is
+refused rather than resolved, because every ambiguity here ends as either "my
+file went to the wrong disk" or "the same film is listed twice and deleting one
+copy breaks the other".
+"""
+
+from pathlib import Path
+
+import pytest
+
+from meshbay_node.roots import Root, RootError, RootSet, entry_abs_path
+from meshbay_common.protocol import IndexEntry
+
+
+def _spec(path, **kw):
+    return {"path": str(path), **kw}
+
+
+def _entry(path: str, name: str) -> IndexEntry:
+    return IndexEntry(id="f" * 64, name=name, path=path, size=1,
+                      type="other", added_at=0)
+
+
+# ── Naming ───────────────────────────────────────────────────────────────────
+
+def test_the_name_is_the_directory_basename(tmp_path):
+    (tmp_path / "Films").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Films")])
+    assert roots.names == ["Films"]
+
+
+def test_an_explicit_name_wins_over_the_basename(tmp_path):
+    (tmp_path / "Films").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Films", name="Cinema")])
+    assert roots.names == ["Cinema"]
+
+
+def test_two_roots_cannot_share_a_name(tmp_path):
+    for parent in ("a", "b"):
+        (tmp_path / parent / "Films").mkdir(parents=True)
+    with pytest.raises(RootError, match="both be called"):
+        RootSet.build([_spec(tmp_path / "a" / "Films"),
+                       _spec(tmp_path / "b" / "Films")])
+
+
+def test_names_clash_without_regard_to_case(tmp_path):
+    """
+    `Films` and `films` are one directory on NTFS and exFAT, which is where most
+    of these live. A comparison that respected case would let the pair through
+    and produce two roots a Windows member cannot tell apart.
+    """
+    (tmp_path / "a" / "Films").mkdir(parents=True)
+    (tmp_path / "b" / "films").mkdir(parents=True)
+    with pytest.raises(RootError, match="both be called"):
+        RootSet.build([_spec(tmp_path / "a" / "Films"),
+                       _spec(tmp_path / "b" / "films")])
+
+
+def test_a_name_windows_cannot_write_is_refused(tmp_path):
+    """
+    The root name is a folder every member sees, including on Windows, where
+    `AUX` cannot be created at all.
+    """
+    (tmp_path / "AUX").mkdir()
+    with pytest.raises(RootError, match="reserved on Windows"):
+        RootSet.build([_spec(tmp_path / "AUX")])
+
+
+# ── Nesting ──────────────────────────────────────────────────────────────────
+
+def test_a_root_inside_another_is_refused(tmp_path):
+    """
+    Both roots would index the same bytes under two identities, and deleting
+    through one would leave the other pointing at nothing.
+    """
+    (tmp_path / "Media" / "Films").mkdir(parents=True)
+    with pytest.raises(RootError, match="is inside root"):
+        RootSet.build([_spec(tmp_path / "Media"),
+                       _spec(tmp_path / "Media" / "Films")])
+
+
+def test_nesting_is_refused_in_either_order(tmp_path):
+    (tmp_path / "Media" / "Films").mkdir(parents=True)
+    with pytest.raises(RootError, match="is inside root"):
+        RootSet.build([_spec(tmp_path / "Media" / "Films"),
+                       _spec(tmp_path / "Media")])
+
+
+def test_the_same_directory_twice_is_refused(tmp_path):
+    (tmp_path / "Media").mkdir()
+    with pytest.raises(RootError, match="same directory"):
+        RootSet.build([_spec(tmp_path / "Media"),
+                       _spec(tmp_path / "Media", name="Other")])
+
+
+def test_a_sibling_with_a_shared_prefix_is_fine(tmp_path):
+    """`/data/Media` and `/data/Media2` are unrelated — a string prefix test
+    would wrongly call the second nested inside the first."""
+    (tmp_path / "Media").mkdir()
+    (tmp_path / "Media2").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media"), _spec(tmp_path / "Media2")])
+    assert roots.names == ["Media", "Media2"]
+
+
+# ── Uploads ──────────────────────────────────────────────────────────────────
+
+def test_a_single_root_receives_uploads_without_being_asked(tmp_path):
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    assert roots.upload_root is roots.roots[0]
+
+
+def test_several_roots_and_no_designation_means_no_uploads(tmp_path):
+    """
+    Refused, never guessed: picking one would send a member's file to a disk the
+    operator did not intend, and that is discovered weeks later.
+    """
+    (tmp_path / "A").mkdir()
+    (tmp_path / "B").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "A"), _spec(tmp_path / "B")])
+    assert roots.upload_root is None
+
+
+def test_two_upload_roots_are_refused(tmp_path):
+    (tmp_path / "A").mkdir()
+    (tmp_path / "B").mkdir()
+    with pytest.raises(RootError, match="exactly one"):
+        RootSet.build([_spec(tmp_path / "A", upload=True),
+                       _spec(tmp_path / "B", upload=True)])
+
+
+# ── Resolution ───────────────────────────────────────────────────────────────
+
+def test_resolution_finds_a_path_inside_its_root(tmp_path):
+    (tmp_path / "Media" / "2024").mkdir(parents=True)
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    assert roots.resolve("Media/2024") == (tmp_path / "Media" / "2024").resolve()
+
+
+def test_the_virtual_root_resolves_to_nothing(tmp_path):
+    """
+    It is not a directory on anyone's disk — it belongs to no volume — so a file
+    cannot be written there and a directory cannot be created there.
+    """
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    for attempt in ("", "/", ".", "   "):
+        assert roots.resolve(attempt) is None, f"{attempt!r} resolved"
+
+
+@pytest.mark.parametrize("attempt", [
+    "Media/../..", "Media/../../etc", "Media/sub/../../../etc",
+    "../Media", "..", "Unknown/x",
+])
+def test_escaping_a_root_is_refused(tmp_path, attempt):
+    (tmp_path / "Media" / "sub").mkdir(parents=True)
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    assert roots.resolve(attempt) is None, f"{attempt!r} escaped its root"
+
+
+def test_a_symlink_out_of_the_root_is_refused(tmp_path):
+    """Resolved before comparing, so a link is followed and then rejected —
+    checking the string would have accepted it."""
+    (tmp_path / "Media").mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    (tmp_path / "Media" / "escape").symlink_to(outside)
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    assert roots.resolve("Media/escape") is None
+
+
+def test_resolution_is_case_insensitive_on_the_root_name(tmp_path):
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    assert roots.resolve("media") == (tmp_path / "Media").resolve()
+
+
+def test_an_unavailable_root_resolves_to_nothing(tmp_path):
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    roots.roots[0].available = False
+    assert roots.resolve("Media") is None
+    # …but the mapping is still known, so entries can still be listed as frozen
+    # rather than vanishing from the index.
+    assert roots.split("Media")[0].name == "Media"
+
+
+def test_an_entry_under_a_missing_root_has_no_path(tmp_path):
+    """
+    An unplugged drive must answer "nowhere", not open a file that happens to
+    share a relative path with another root.
+    """
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    entry = _entry("Media", "film.mkv")
+    assert entry_abs_path(roots, entry) == (tmp_path / "Media" / "film.mkv").resolve()
+    roots.roots[0].available = False
+    assert entry_abs_path(roots, entry) is None
+
+
+def test_virtual_path_round_trips(tmp_path):
+    (tmp_path / "Media" / "2024").mkdir(parents=True)
+    roots = RootSet.build([_spec(tmp_path / "Media")])
+    real = roots.resolve("Media/2024")
+    assert roots.virtual_of(real) == "Media/2024"
+    assert roots.virtual_of(roots.resolve("Media")) == "Media"
+
+
+# ── Availability ─────────────────────────────────────────────────────────────
+
+def test_availability_follows_the_directory(tmp_path):
+    target = tmp_path / "Media"
+    target.mkdir()
+    roots = RootSet.build([_spec(target)])
+    assert roots.refresh_availability() == []
+
+    target.rmdir()                       # stands in for an unmounted volume
+    changed = roots.refresh_availability()
+    assert [(r.name, live) for r, live in changed] == [("Media", False)]
+    assert roots.roots[0].available is False
+
+    target.mkdir()
+    changed = roots.refresh_availability()
+    assert [(r.name, live) for r, live in changed] == [("Media", True)]
+
+
+def test_describe_reports_what_a_member_needs(tmp_path):
+    (tmp_path / "Media").mkdir()
+    (tmp_path / "Music").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", upload=True),
+                           _spec(tmp_path / "Music", kind="audio")])
+    described = roots.describe()
+    assert described == [
+        {"name": "Media", "kind": "generic", "available": True, "upload": True},
+        {"name": "Music", "kind": "audio", "available": True, "upload": False},
+    ]
+    # Deliberately no paths: a member is told what exists and whether it is
+    # readable, not where on the operator's disk it lives.
+    assert not any("path" in d for d in described)
diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py
index 435cc76..9e45dbc 100644
--- a/packages/meshbay-node/tests/test_roster_pairing.py
+++ b/packages/meshbay-node/tests/test_roster_pairing.py
@@ -22,6 +22,7 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from meshbay_common.crypto import generate_gek, pk_to_b64, unwrap_gek_aes
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
 from meshbay_node.indexer.group_index import GroupIndex
+from conftest import one_root
 from meshbay_node.roster import Roster, hash_code, normalize_code
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
@@ -65,7 +66,7 @@ def _session(tmp_path: Path, roster, user_id: str = "grenet",
 
     session = WebRTCPeerSession.__new__(WebRTCPeerSession)
     session._ctx = {
-        "shared_root": shared_root,
+        "roots": one_root(shared_root),
         "index": index,
         "sk_node": index.sk_node,
         "roster": roster,
@@ -74,7 +75,7 @@ def _session(tmp_path: Path, roster, user_id: str = "grenet",
         session._ctx["groups"] = {
             group_id: {
                 "gek": gek,
-                "shared_root": shared_root,
+                "roots": one_root(shared_root),
                 "index": index,
                 "join_policy": join_policy,
             },
@@ -599,7 +600,7 @@ async def test_revoke_endpoint_stops_authorization(tmp_path, roster):
 
     resp = client.post(f"/api/members/bob/revoke?group_id={GROUP}&t=tok")
     assert resp.status_code == 200
-    assert "gek-init" in resp.json()["reminder"], (
+    assert "gek rotate" in resp.json()["reminder"], (
         "revocation must remind the operator to rotate the key they still hold")
     assert not await roster.is_authorized(GROUP, "bob")
 
@@ -823,7 +824,7 @@ async def test_a_directory_with_anything_in_it_is_refused(tmp_path, roster):
     full.mkdir()
     (full / "keep.txt").write_text("still here")
 
-    session._do_dir_delete({"dir": "full"})
+    session._do_dir_delete({"dir": "shared/full"})
 
     assert _last(session).get("detail") == "Directory is not empty"
     assert full.exists() and (full / "keep.txt").exists()
@@ -835,15 +836,23 @@ async def test_no_challenge_is_issued_without_an_operator(tmp_path, roster):
     session._ctx["has_admin_authority"] = False
     (tmp_path / "shared" / "empty").mkdir()
 
-    session._do_dir_delete({"dir": "empty"})
+    session._do_dir_delete({"dir": "shared/empty"})
 
     assert _last(session).get("detail") == "No authorized key for deletion"
     assert (tmp_path / "shared" / "empty").exists()
 
 
-async def test_the_shared_root_itself_is_not_a_target(tmp_path, roster):
+async def test_a_root_itself_is_not_a_target(tmp_path, roster):
+    """
+    Neither the virtual root nor a root directory can be removed this way.
+
+    Removing a root is a configuration change: doing it through a file operation
+    would leave the group config naming a directory nobody can reach. And the
+    virtual root is not a directory on anyone's disk at all — it belongs to no
+    volume.
+    """
     session = await _dir_session(tmp_path, roster)
-    for attempt in ("", ".", "/", "../shared"):
+    for attempt in ("", ".", "/", "../shared", "shared", "shared/", "SHARED"):
         session._do_dir_delete({"dir": attempt})
         assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
     assert (tmp_path / "shared").is_dir()
@@ -854,7 +863,11 @@ async def test_escaping_the_shared_root_is_refused(tmp_path, roster):
     outside = tmp_path / "outside"
     outside.mkdir()
 
-    for attempt in ("../outside", "../../outside", "sub/../../outside"):
+    # Both shapes: a path that names no root at all, and one that starts inside
+    # a real root and then climbs out of it.
+    for attempt in ("../outside", "../../outside", "sub/../../outside",
+                    "shared/../outside", "shared/../../outside",
+                    "shared/sub/../../outside"):
         session._do_dir_delete({"dir": attempt})
         assert _last(session).get("type") == "error", f"{attempt!r} was accepted"
     assert outside.is_dir(), "a path leaving the shared root removed a directory"
@@ -871,19 +884,19 @@ async def test_an_empty_directory_needs_a_signature_and_then_goes(tmp_path, rost
     session = await _dir_session(tmp_path, roster)
     (tmp_path / "shared" / "gone").mkdir()
 
-    session._do_dir_delete({"dir": "gone"})
+    session._do_dir_delete({"dir": "shared/gone"})
     challenge = _last(session)
     assert challenge["type"] == "admin_challenge"
     assert challenge["op"] == "dir_delete"
-    assert challenge["subject"] == "gone"
+    assert challenge["subject"] == "shared/gone"
 
     transcript = admin_transcript(
         op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1",
-        subject="gone", nonce=base64.b64decode(challenge["nonce"]),
+        subject="shared/gone", nonce=base64.b64decode(challenge["nonce"]),
         ts=challenge["ts"])
     await session._admin_exec_dir_delete(
         session._admin_ops.pop(challenge["op_id"]) if session._admin_ops
-        else {"op": "dir_delete", "subject": "gone"},
+        else {"op": "dir_delete", "subject": "shared/gone"},
         transcript, sk_ed.sign(transcript))
 
     assert _last(session)["type"] == "dir_delete_ack"
@@ -906,9 +919,9 @@ async def test_someone_elses_signature_does_not_remove_it(tmp_path, roster):
 
     transcript = admin_transcript(
         op="dir_delete", node_pk_b64=session._node_pk_b64(), group_id="g1",
-        subject="theirs", nonce=b"\x22" * 32, ts=int(time.time()))
+        subject="shared/theirs", nonce=b"\x22" * 32, ts=int(time.time()))
     await session._admin_exec_dir_delete(
-        {"op": "dir_delete", "subject": "theirs"},
+        {"op": "dir_delete", "subject": "shared/theirs"},
         transcript, sk_member.sign(transcript))
 
     assert _last(session).get("detail") == "Signature verification failed"
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index e13dec0..78a631a 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -18,6 +18,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.group_index import GroupIndex
+from conftest import one_root
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
 
@@ -129,12 +130,24 @@ def test_the_node_never_generates_a_name_it_would_refuse(tmp_path):
         f"the node picked {chosen!r} and would then reject it on the next upload")
 
 
+def _uploads_dir(session) -> Path:
+    """
+    Where this session's uploads land: uploads/ inside the group's upload root.
+
+    Asked of the root set rather than assembled by hand, so a test cannot pass
+    while agreeing with a wrong answer the code also produced.
+    """
+    root = session._ctx["roots"].upload_root
+    assert root is not None, "the fixture must designate an upload root"
+    return root.path / "uploads"
+
+
 def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
     """A peer session wired to a real shared root, with sending stubbed out."""
     shared_root = tmp_path / "shared"
     shared_root.mkdir(exist_ok=True)
     index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
-    ctx = {"shared_root": shared_root, "index": index, "sk_node": index.sk_node}
+    ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node}
 
     session = WebRTCPeerSession.__new__(WebRTCPeerSession)
     session._ctx = ctx
@@ -161,9 +174,7 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
     this test now asserts — an existing file is never replaced.
     """
     victim = _session(tmp_path, "victim-user")
-    shared_root = victim._ctx["shared_root"]
-
-    uploads = shared_root / "uploads"
+    uploads = _uploads_dir(victim)
     uploads.mkdir()
     original = uploads / "important.mp4"
     original.write_bytes(b"operator's original content")
@@ -190,7 +201,7 @@ def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
     session.sent.clear()
 
     session._do_file_upload(dict(payload))
-    uploads = session._ctx["shared_root"] / "uploads"
+    uploads = _uploads_dir(session)
     assert (uploads / "movie.mp4").read_bytes() == b"first", (
         "the first upload was replaced")
     assert (uploads / "movie (2).mp4").read_bytes() == b"first"
@@ -221,7 +232,6 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
     client-chosen destination would open does not exist on this path.
     """
     session = _session(tmp_path, "user-1")
-    shared_root = session._ctx["shared_root"]
 
     session._do_file_upload({
         "filename": "note.txt", "dir": "../../etc",
@@ -229,7 +239,7 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
         "data": base64.b64encode(b"x").decode(),
     })
 
-    assert (shared_root / "uploads" / "note.txt").read_bytes() == b"x"
+    assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x"
     assert not (tmp_path / "etc").exists()
 
 
@@ -249,7 +259,7 @@ def test_two_members_can_send_the_same_filename(tmp_path):
         "data": base64.b64encode(b"second").decode(),
     })
 
-    uploads = first._ctx["shared_root"] / "uploads"
+    uploads = _uploads_dir(first)
     assert (uploads / "IMG_1234.jpg").read_bytes() == b"first"
     assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second"
 
@@ -270,8 +280,8 @@ def test_chat_store_and_peers_are_per_group(tmp_path):
     index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate())
     index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate())
     groups = {
-        "a" * 32: {"chat_store": "STORE_A", "index": index_a, "shared_root": tmp_path},
-        "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path},
+        "a" * 32: {"chat_store": "STORE_A", "index": index_a, "roots": one_root(tmp_path / "a")},
+        "b" * 32: {"chat_store": "STORE_B", "index": index_b, "roots": one_root(tmp_path / "b")},
     }
     ctx = {"groups": groups}
 
@@ -663,7 +673,7 @@ def test_admin_ui_escapes_filenames(tmp_path):
 
     html = _render_page({
         "status": "running",
-        "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}},
+        "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(tmp_path)}},
         "indexes": {"g" * 32: index},
     })
 
diff --git a/packages/meshbay-node/tests/test_stream_capacity_config.py b/packages/meshbay-node/tests/test_stream_capacity_config.py
index 7c33a2f..cf90e22 100644
--- a/packages/meshbay-node/tests/test_stream_capacity_config.py
+++ b/packages/meshbay-node/tests/test_stream_capacity_config.py
@@ -25,6 +25,7 @@ from pathlib import Path
 import pytest
 
 from meshbay_node.config import load_config
+from meshbay_node.roots import RootSet
 from meshbay_node.transport.webrtc_server import (
     MAX_CONCURRENT_TRANSCODES,
     WebRTCPeerSession,
@@ -95,7 +96,7 @@ class _FakePC:
 def _semaphore_size(n):
     t = WebRTCTransport(
         sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
-        shared_root=Path("/tmp"), index=None, max_concurrent_streams=n)
+        roots=RootSet(), index=None, max_concurrent_streams=n)
     s = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="p")
     return s._transcode_semaphore()._value
 
@@ -117,7 +118,7 @@ def test_the_budget_is_shared_between_peers():
     """
     t = WebRTCTransport(
         sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
-        shared_root=Path("/tmp"), index=None, max_concurrent_streams=2)
+        roots=RootSet(), index=None, max_concurrent_streams=2)
 
     async def go():
         a = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="a")
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index cc0c6a2..5727ef9 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -48,6 +48,7 @@ from meshbay_common.adminop import (
 )
 from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR, join_transcript
 from meshbay_node.bundle_store import BundleStore
+from conftest import one_root
 from meshbay_node.roster import Roster
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.transport.webrtc_server import WebRTCTransport
@@ -265,12 +266,12 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us
 async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: browser sends MNP handshake, node responds with handshake_ack."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -309,12 +310,12 @@ async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir):
 async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: full file transfer — handshake, index, fetch chunk, decrypt."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -396,12 +397,12 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir
 async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: invalid JWT is rejected with error."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -444,12 +445,12 @@ async def test_webrtc_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir):
 async def test_webrtc_request_before_handshake_rejected(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: request without handshake is rejected."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -490,7 +491,7 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm
     from meshbay_node.chat.store import ChatStore
 
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     chat_store = ChatStore(db_path=tmp_path / "chat_test.db")
@@ -498,7 +499,7 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["chat_store"] = chat_store
@@ -537,12 +538,12 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm
 async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: chat history without chat_store returns empty list."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -566,7 +567,7 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path)
     from meshbay_node.chat.store import ChatStore
 
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     chat_store = ChatStore(db_path=tmp_path / "chat_bc.db")
@@ -574,7 +575,7 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path)
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["chat_store"] = chat_store
@@ -604,12 +605,12 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path)
 async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: JWT without matching group claim is rejected."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -651,12 +652,12 @@ async def test_webrtc_group_membership_enforced(sk_node, sk_hub, gek, shared_dir
 async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: peer removed from _peers dict on session close."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -678,12 +679,12 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
 async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: stream_segment for non-existent file returns error."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -708,12 +709,12 @@ async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_d
 async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -768,12 +769,12 @@ async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir)
 async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, shared_dir):
     """WebRTC: DTLS channel binding detects fingerprint substitution (simulated MitM)."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -850,14 +851,14 @@ async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir,
                                               tmp_path):
     """WebRTC DataChannel: admin file delete requires Ed25519 challenge-response."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     sk_admin = Ed25519PrivateKey.generate()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["roster"] = await _paired_operator_roster(tmp_path, sk_admin)
@@ -899,7 +900,7 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_
                                                   tmp_path):
     """WebRTC DataChannel: wrong Ed25519 signature is rejected — hub can't fake admin."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     sk_admin = Ed25519PrivateKey.generate()
@@ -907,7 +908,7 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["roster"] = await _paired_operator_roster(tmp_path, sk_admin)
@@ -946,12 +947,12 @@ async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_
 async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: stream_request for non-existent file returns error."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -975,7 +976,7 @@ async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_d
 async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, shared_dir):
     """Uploader must prove Ed25519 key ownership to delete — no uploader shortcut."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     sk_uploader = Ed25519PrivateKey.generate()
@@ -986,7 +987,7 @@ async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, s
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     # No admin_pk configured — only uploader_pk should authorize deletion
@@ -1032,7 +1033,7 @@ async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, s
 async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, shared_dir):
     """Hub-forged JWT with same sub cannot delete — wrong Ed25519 key is rejected."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     # User A uploaded the file
@@ -1047,7 +1048,7 @@ async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, share
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     # No admin_pk — only uploader_pk matters
@@ -1117,7 +1118,7 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di
     that lookup was H3.
     """
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     roster = Roster(db_path=tmp_path / "roster.db")
@@ -1125,13 +1126,13 @@ async def test_invite_then_join_delivers_the_gek(sk_node, sk_hub, gek, shared_di
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["roster"] = roster
     transport._ctx["has_admin_authority"] = True
     transport._ctx["groups"] = {
-        TEST_GROUP: {"gek": gek, "shared_root": shared_dir, "index": indexer.index},
+        TEST_GROUP: {"gek": gek, "roots": shared_dir, "index": indexer.index},
     }
 
     # A paired operator, as `meshbay-node operator pair` would have left it.
@@ -1229,7 +1230,7 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
                                                   tmp_path, x25519_keypair):
     """Browser fetches GEK bundle from node during the handshake challenge window."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     sk_x_raw, pk_x_raw = x25519_keypair
@@ -1244,7 +1245,7 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["bundle_store"] = bundle_store
@@ -1328,7 +1329,7 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
 async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, tmp_path):
     """Keypair bundle stored on node, then fetched during handshake window."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
@@ -1336,7 +1337,7 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["bundle_store"] = bundle_store
@@ -1415,7 +1416,7 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
 async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path):
     """Keypair bundle fetch returns found=false when no bundle exists."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
@@ -1423,7 +1424,7 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir,
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["bundle_store"] = bundle_store
@@ -1492,7 +1493,7 @@ async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shar
     through the node's local admin UI or CLI.
     """
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     sk_x_raw, pk_x_raw = x25519_keypair
@@ -1504,7 +1505,7 @@ async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shar
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["bundle_store"] = bundle_store
@@ -1547,12 +1548,12 @@ async def test_gek_not_auto_activated_on_bundle_store(sk_node, sk_hub, gek, shar
 async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir):
     """WebRTC DataChannel: connection refused when GEK is not initialized."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=None)
     await indexer.initial_scan()
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=None,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
 
@@ -1594,7 +1595,7 @@ async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir):
 async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path):
     """GEK bundle fetch returns found=false when no bundle exists."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
-    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
     bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
@@ -1602,7 +1603,7 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_
 
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
-        shared_root=shared_dir, index=indexer.index,
+        roots=one_root(shared_dir), index=indexer.index,
         stun_servers=[],
     )
     transport._ctx["bundle_store"] = bundle_store