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>

14 files changed, 1289 insertions, 101 deletions

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