Merge branch 'refactor/groups-phase1'

Groups refactor, phases 1-3.

The root model replaces the old `upload` flag and group-wide `member_upload`
with per-root `writable`/`removable`/`ejected`, carried by a `RootSet` that
both front doors — the loopback API and signed MNP — reach through the same
`ops` functions. MNP goes to 1.1, additively: the roots table now rides on
`index_delta`, so a root added, removed, ejected or plugged reaches every
connected client instead of only whoever reloaded.

The group UI becomes a plugin architecture: an application is a registry
entry in `apps.js` plus its own files, with directories stored generically
by `ops.set_app_directories` under whatever the app is called. A reference
application, hidden behind `?dev=1`, is what makes that claim testable —
adding it is what found the two places still naming apps by hand.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011pvMdvLBG92jyhvD5pD6us

24 files changed, 2454 insertions, 265 deletions

diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py
index 20724aa..3dc9cd9 100644
--- a/packages/meshbay-node/tests/conftest.py
+++ b/packages/meshbay-node/tests/conftest.py
@@ -24,14 +24,18 @@ win32_todo = pytest.mark.skipif(
 )
 
 
-def one_root(path: Path, *, name: str = "", kind: str = "generic") -> RootSet:
+def one_root(path: Path, *, name: str = "", kind: str = "generic",
+             writable: bool = True) -> RootSet:
     """
-    A RootSet with a single root over `path`, receiving uploads.
+    A RootSet with a single writable root over `path`.
 
     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.
+
+    Writable by default because most callers are testing something else and
+    want a root an upload can reach. `writable=False` is the read-only group.
     """
     return RootSet.build([{"path": str(path), "name": name, "kind": kind,
-                           "upload": True}])
+                           "writable": writable}])
diff --git a/packages/meshbay-node/tests/test_app_directories.py b/packages/meshbay-node/tests/test_app_directories.py
new file mode 100644
index 0000000..3ede1b6
--- /dev/null
+++ b/packages/meshbay-node/tests/test_app_directories.py
@@ -0,0 +1,292 @@
+"""
+One shape for every application's directories.
+
+`video_root` (a string), `audio_root` (a string) and `photo_roots` (a list)
+said the same thing three ways, and each needed its own op, its own MNP message
+and its own settings widget. They are one function keyed by the app's own name
+now, which is what lets an application be added without touching this layer at
+all — the whole claim of the plugin architecture.
+
+Two properties are new rather than moved, and both matter more than the tidying:
+
+* **the paths are validated.** The setters this replaces accepted anything. A
+  typo, or a path left behind when a root was removed, was stored happily and
+  then matched no entry — an app showing an empty tab, with nothing to
+  distinguish "misconfigured" from "no files yet". The moment of setting is the
+  only one where the operator is present to be told;
+* **the legacy scalar is derived, never stored.** `video_root` still rides on
+  the handshake ack for MNP 1.0 clients. Kept as a second stored value it would
+  drift from the list within one run — the shape of bug that reads as "it works
+  after a restart".
+"""
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+async def _state(tmp_path: Path, *, writable: bool = True) -> tuple[dict, Roster]:
+    media = tmp_path / "Media"
+    (media / "Films").mkdir(parents=True)
+    (media / "Albums").mkdir()
+    published = tmp_path / "Published"
+    published.mkdir()
+
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roots = RootSet.build([
+        {"path": str(media), "writable": writable},
+        {"path": str(published)},
+    ])
+    state = {
+        "roster": roster,
+        "node_user_id": "operator",
+        "groups_ctx": {GROUP: {"index": index, "roots": roots}},
+        "config": SimpleNamespace(groups=[SimpleNamespace(id=GROUP, roots=[])]),
+    }
+    return state, roster
+
+
+# ── One function, any app ────────────────────────────────────────────────────
+
+async def test_an_app_nobody_wrote_code_for_stores_its_directories(tmp_path):
+    """
+    The point of the generic pair. Nothing in ops.py, roster.py or the daemon
+    names this app, and it round-trips anyway — which is the difference between
+    a plugin architecture and a list of special cases.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        await ops.set_app_directories(state, GROUP, "helloworld", ["Media/Films"])
+        assert await roster.app_directories(GROUP, "helloworld") == ["Media/Films"]
+    finally:
+        await roster.close()
+
+
+async def test_directories_are_deduplicated_and_ordered(tmp_path):
+    """
+    The stored form is what the operator signs a subject built from, on both
+    sides. Two clients sending the same set in different orders must produce
+    the same bytes, or one of them refuses to sign its own request.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        out = await ops.set_app_directories(
+            state, GROUP, "video", ["Media/Films", "Media", "Media/Films"])
+        assert out["directories"] == ["Media", "Media/Films"]
+    finally:
+        await roster.close()
+
+
+async def test_a_single_directory_app_stores_a_one_element_list(tmp_path):
+    state, roster = await _state(tmp_path)
+    try:
+        out = await ops.set_app_directory(state, GROUP, "chat", "Media/Films",
+                                          require_writable=True)
+        assert out["path"] == "Media/Films"
+        assert await roster.app_directories(GROUP, "chat") == ["Media/Films"]
+
+        cleared = await ops.set_app_directory(state, GROUP, "chat", "")
+        assert cleared["path"] == ""
+        assert await roster.app_directories(GROUP, "chat") == []
+    finally:
+        await roster.close()
+
+
+# ── Validation ───────────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("bad", [
+    "Nowhere", "Nowhere/Deeper", "/etc", "Media/../../etc", "..",
+])
+async def test_a_directory_outside_every_root_is_refused(tmp_path, bad):
+    state, roster = await _state(tmp_path)
+    try:
+        with pytest.raises(ops.OpError):
+            await ops.set_app_directories(state, GROUP, "video", [bad])
+        assert await roster.app_directories(GROUP, "video") == []
+    finally:
+        await roster.close()
+
+
+async def test_a_read_only_root_is_refused_where_writability_is_required(tmp_path):
+    """
+    Chat's directory is a destination, not a view. Storing one on a read-only
+    root would produce a paperclip that fails at the moment somebody uses it,
+    which is the failure the RO/RW model exists to move earlier.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        with pytest.raises(ops.OpError, match="read-only"):
+            await ops.set_app_directory(state, GROUP, "chat", "Published",
+                                        require_writable=True)
+        # The same path is fine for an app that only reads it.
+        await ops.set_app_directories(state, GROUP, "video", ["Published"])
+        assert await roster.app_directories(GROUP, "video") == ["Published"]
+    finally:
+        await roster.close()
+
+
+async def test_a_directory_on_an_unplugged_drive_can_still_be_configured(tmp_path):
+    """
+    Deliberately *not* `RootSet.resolve()`, which also refuses a root that is
+    currently unavailable. An operator must be able to point an app at a
+    library on a drive they have ejected — what is checked is the shape, which
+    does not change with what happens to be mounted.
+    """
+    state, roster = await _state(tmp_path)
+    roots = state["groups_ctx"][GROUP]["roots"]
+    roots.roots[0].available = False
+    try:
+        out = await ops.set_app_directories(state, GROUP, "video", ["Media/Films"])
+        assert out["directories"] == ["Media/Films"]
+    finally:
+        await roster.close()
+
+
+# ── The derived scalar ───────────────────────────────────────────────────────
+
+async def test_the_legacy_scalar_follows_the_list_in_the_live_context(tmp_path):
+    """
+    `video_root` rides on the handshake ack for MNP 1.0 clients and is read
+    from the group context. Left behind by a save, it would disagree with the
+    list until the next restart.
+    """
+    state, roster = await _state(tmp_path)
+    ctx = state["groups_ctx"][GROUP]
+    try:
+        await ops.set_app_directories(state, GROUP, "video",
+                                      ["Media/Films", "Media/Albums"])
+        assert ctx["video_directories"] == ["Media/Albums", "Media/Films"]
+        assert ctx["video_root"] == "Media/Albums", (
+            "the scalar must be the first of the list, not a stale value")
+
+        await ops.set_app_directories(state, GROUP, "video", [])
+        assert ctx["video_root"] == ""
+    finally:
+        await roster.close()
+
+
+async def test_the_photo_alias_stays_a_list_and_chat_stays_a_string(tmp_path):
+    """The alias table has to carry the shape, not just the name."""
+    state, roster = await _state(tmp_path)
+    ctx = state["groups_ctx"][GROUP]
+    try:
+        await ops.set_app_directories(state, GROUP, "photo",
+                                      ["Media/Films", "Media/Albums"])
+        assert ctx["photo_roots"] == ["Media/Albums", "Media/Films"]
+        await ops.set_app_directory(state, GROUP, "chat", "Media",
+                                    require_writable=True)
+        assert ctx["chat_directory"] == "Media"
+    finally:
+        await roster.close()
+
+
+# ── Reading what an older node stored ────────────────────────────────────────
+
+async def test_an_existing_video_root_is_read_without_a_migration(tmp_path):
+    """
+    A node upgraded into this reads its old key until the first save through
+    the new path. Requiring a migration script to run before the Videos tab
+    works again would be a step nobody performs on the machine where it
+    matters.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "Media/Films")
+        assert await roster.app_directories(GROUP, "video") == ["Media/Films"]
+
+        await roster.set_setting(GROUP, "audio_root", "Media/Albums")
+        assert await roster.app_directories(GROUP, "music") == ["Media/Albums"]
+
+        await roster.set_setting(GROUP, "photo_roots", '["A", "B"]')
+        assert await roster.app_directories(GROUP, "photo") == ["A", "B"]
+    finally:
+        await roster.close()
+
+
+async def test_an_empty_legacy_value_means_nothing_configured(tmp_path):
+    """`video_root = ""` was how "unset" was spelled; it must not become `[""]`."""
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "")
+        assert await roster.app_directories(GROUP, "video") == []
+    finally:
+        await roster.close()
+
+
+async def test_the_new_key_wins_over_the_legacy_one(tmp_path):
+    """
+    Both present is a group saved once through the new path. Reading the old
+    key there would undo that save on every load.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_setting(GROUP, "video_root", "Old/Place")
+        await roster.set_app_directories(GROUP, "video", ["New/Place"])
+        assert await roster.app_directories(GROUP, "video") == ["New/Place"]
+    finally:
+        await roster.close()
+
+
+async def test_an_app_with_no_legacy_name_simply_has_none(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.app_directories(GROUP, "helloworld") == []
+    finally:
+        await roster.close()
+
+
+# ── Chat's own settings ──────────────────────────────────────────────────────
+
+async def test_link_previews_default_on_and_survive_a_restart(tmp_path):
+    """
+    Absent means on, because that is what the node did before the switch
+    existed — an upgrade must not silently change what a group's chat does.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.chat_link_preview(GROUP) is True
+        await roster.set_chat_link_preview(GROUP, False, set_by="op")
+        assert await roster.chat_link_preview(GROUP) is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.chat_link_preview(GROUP) is False
+        assert await reopened.chat_link_preview("other") is True, (
+            "one group's setting must not answer for another")
+    finally:
+        await reopened.close()
+
+
+async def test_setting_link_previews_updates_the_live_context(tmp_path):
+    """
+    The unfurl handler reads the context, not the database — it runs per
+    message and a round trip there would be absurd. So the two are kept in
+    step by the op that changes it.
+    """
+    state, roster = await _state(tmp_path)
+    try:
+        await ops.set_chat_link_preview(state, GROUP, False)
+        assert state["groups_ctx"][GROUP]["chat_link_preview"] is False
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_apps_enabled_policy.py b/packages/meshbay-node/tests/test_apps_enabled_policy.py
index 671005a..ac44ab3 100644
--- a/packages/meshbay-node/tests/test_apps_enabled_policy.py
+++ b/packages/meshbay-node/tests/test_apps_enabled_policy.py
@@ -1,7 +1,7 @@
 """
 The operator decides which group "applications" (Chat, Files, ...) are shown.
 
-Same shape as `test_member_upload_policy.py`, because it is the same kind of
+Same shape as `test_root_writable_policy.py`, because it is the same kind of
 setting: changed by a signed operator instruction, stored on the node rather
 than the hub, and safe for an existing group to have never heard of. The two
 things specific to this one: the whole set is signed in one message rather
@@ -88,7 +88,7 @@ async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
 
 async def test_changing_it_needs_a_signature(tmp_path):
     """The request only ever produces a challenge. Nothing is applied until a
-    signature over the transcript verifies — the same path as member_upload."""
+    signature over the transcript verifies — the same path as the root ops."""
     session = _session(tmp_path, "op", operator="op")
     session._has_admin_authority = lambda: True
     issued = []
diff --git a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
index deab1bd..5a8f9b1 100644
--- a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
+++ b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py
@@ -102,7 +102,7 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
 
     daemon = await _make_daemon(tmp_path, shared, group_id)
     try:
-        await daemon._roster.set_audio_root(group_id, "shared/Music", set_by="op")
+        await daemon._roster.set_app_directories(group_id, "music", ["shared/Music"], set_by="op")
         indexer = DirectoryIndexer(
             roots=one_root(shared), group_id=group_id,
             sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
@@ -142,8 +142,10 @@ async def test_setting_the_audio_root_sweeps_what_it_already_contains(tmp_path):
         # this file all along.
         state = {
             "roster": daemon._roster,
-            "groups_ctx": {group_id: {}},
-            "enrich_audio_root_fn": daemon._enrich_audio_root_now,
+            # Real roots, because ops now refuses a directory that is not
+            # inside one — the per-app setters this replaced validated nothing.
+            "groups_ctx": {group_id: {"roots": one_root(shared)}},
+            "enrich_app_dirs_fns": {"music": daemon._enrich_audio_root_now},
         }
         await ops.set_audio_root(state, group_id, "shared/Music")
         await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
@@ -197,8 +199,10 @@ async def test_the_same_file_shared_into_two_groups_enriches_in_both(tmp_path):
             "the fixture itself must produce identical content hashes — "
             "otherwise this test isn't exercising the collision at all")
 
-        await daemon._roster.set_audio_root(group_a, "shared_a/Music", set_by="op")
-        await daemon._roster.set_audio_root(group_b, "shared_b/Music", set_by="op")
+        await daemon._roster.set_app_directories(
+            group_a, "music", ["shared_a/Music"], set_by="op")
+        await daemon._roster.set_app_directories(
+            group_b, "music", ["shared_b/Music"], set_by="op")
 
         await daemon._enrich_new_audio_entries(indexer_a, list(indexer_a.index.entries))
         await daemon._enrich_new_audio_entries(indexer_b, list(indexer_b.index.entries))
diff --git a/packages/meshbay-node/tests/test_audio_root_policy.py b/packages/meshbay-node/tests/test_audio_root_policy.py
index 576c08a..e2e9254 100644
--- a/packages/meshbay-node/tests/test_audio_root_policy.py
+++ b/packages/meshbay-node/tests/test_audio_root_policy.py
@@ -125,17 +125,20 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        assert await roster.audio_root("g1") == "", "absent must mean unset"
-        await roster.set_audio_root("g1", "shared/Music", set_by="op")
-        assert await roster.audio_root("g1") == "shared/Music"
+        assert await roster.app_directories("g1", "music") == [], (
+            "absent must mean unset")
+        await roster.set_app_directories("g1", "music", ["shared/Music"],
+                                         set_by="op")
+        assert await roster.app_directories("g1", "music") == ["shared/Music"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.audio_root("g1") == "shared/Music"
-        assert await reopened.audio_root("g2") == "", "one group's setting must not answer for another"
+        assert await reopened.app_directories("g1", "music") == ["shared/Music"]
+        assert await reopened.app_directories("g2", "music") == [], (
+            "one group's setting must not answer for another")
     finally:
         await reopened.close()
 
@@ -239,7 +242,7 @@ async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_
 
         assert session._ctx["groups"][GROUP]["audio_root"] == "shared/Music", (
             "the live in-memory context must reflect the new root immediately")
-        assert await roster.audio_root(GROUP) == "shared/Music", (
+        assert await roster.app_directories(GROUP, "music") == ["shared/Music"], (
             "the same Roster instance must read back what it just wrote")
     finally:
         await roster.close()
@@ -249,7 +252,7 @@ async def test_a_real_signed_save_persists_and_survives_a_fresh_roster_read(tmp_
     # actually answers "does it survive a reload".
     reopened = await open_roster(tmp_path)
     try:
-        assert await reopened.audio_root(GROUP) == "shared/Music", (
+        assert await reopened.app_directories(GROUP, "music") == ["shared/Music"], (
             "a freshly-opened Roster against the same db file must see the "
             "committed value — anything else means the write was never "
             "durable in the first place")
diff --git a/packages/meshbay-node/tests/test_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
index f58c020..cf91564 100644
--- a/packages/meshbay-node/tests/test_cli_dispatch.py
+++ b/packages/meshbay-node/tests/test_cli_dispatch.py
@@ -26,6 +26,15 @@ VERBS = [
     ["status"],
     ["group", "list"],
     ["group", "add"],                      # missing --dir: usage, then exit
+    ["group", "add", "g", "--dir", "/tmp/media", "--no-writable"],
+    ["root", "list"],
+    ["root", "add"],                       # missing path: usage, then exit
+    ["root", "add", "/tmp/media", "--writable", "--removable"],
+    ["root", "remove", "media", "--yes"],
+    ["root", "set", "media", "--no-writable"],
+    ["root", "set", "media"],              # nothing to change: usage, then exit
+    ["root", "eject", "media"],
+    ["root", "plug", "media"],
     ["gek", "init"],
     ["gek", "rotate", "--yes"],
     ["gek-init"],
@@ -33,6 +42,9 @@ VERBS = [
     ["member", "invite", "bob"],
     ["member", "revoke", "bob"],
     ["member", "unpin", "bob"],
+    # Removed, and it has to say so rather than offering a username for a verb
+    # that no longer takes one.
+    ["member", "upload"],
     ["operator", "pair"],
     ["file", "list"],
     ["file", "rm", "abc", "--yes"],
@@ -224,3 +236,57 @@ def test_a_bare_invocation_with_no_config_yet_exits_cleanly(monkeypatch, tmp_pat
     assert "meshbay-node init" in capsys.readouterr().out
     assert not missing_config.parent.exists(), (
         "a fresh, unprovisioned start must not create anything on disk")
+
+
+def test_a_removed_verb_says_what_replaced_it():
+    """
+    `member upload` used to set a group-wide switch that no longer exists. It
+    reached the usage line for the *other* member verbs — "usage: meshbay-node
+    member upload <username>" — which advertises a removed feature and sends
+    the operator looking for a username it would then reject.
+
+    Naming it costs three lines and is the difference between an operator
+    finding `root set --writable` and concluding the CLI is broken.
+    """
+    import inspect
+    source = inspect.getsource(daemon_mod.main)
+    start = source.index('if args.command == "member":')
+    block = source[start:source.index('if args.command == "group":', start)]
+
+    assert 'sub == "upload"' in block, (
+        "`member upload` falls through to the generic usage line")
+    guidance = block[block.index('sub == "upload"'):]
+    guidance = guidance[:guidance.index("sys.exit")]
+    assert "root set" in guidance and "--writable" in guidance, (
+        "the message does not name what replaced it")
+
+
+def test_there_is_no_way_to_create_a_group_in_the_old_shape():
+    """
+    `--upload-dir` is gone, and documenting it as deprecated was the wrong
+    answer — which is what it got at first.
+
+    It wrote `upload_dir` into a brand-new `[[groups]]` block, and
+    `GroupConfig.__post_init__` reads that by forcing *every other root
+    read-only* and appending that path as the one writable one. So
+    `group add --dir X --writable --upload-dir Y` silently made X read-only:
+    two mechanisms deciding which directories accept uploads, one of them
+    invisible, in a group created after the model that replaced it.
+
+    Reading it stays — an existing node.toml must keep working, and that is the
+    only legitimate use. Writing it does not.
+    """
+    import inspect
+    source = inspect.getsource(daemon_mod.main)
+    assert "--upload-dir" not in source, (
+        "the CLI can still create a group in the pre-RO/RW shape")
+
+    from meshbay_node import ops
+    params = inspect.signature(ops.attach_group).parameters
+    assert "upload_dir" not in params, (
+        "attach_group still writes the legacy key")
+
+    # The read path is deliberately untouched.
+    from meshbay_node.config import GroupConfig
+    assert "upload_dir" in inspect.getsource(GroupConfig), (
+        "an existing node.toml using upload_dir would stop working")
diff --git a/packages/meshbay-node/tests/test_index_delta_carries_roots.py b/packages/meshbay-node/tests/test_index_delta_carries_roots.py
new file mode 100644
index 0000000..227f2d0
--- /dev/null
+++ b/packages/meshbay-node/tests/test_index_delta_carries_roots.py
@@ -0,0 +1,131 @@
+"""
+The message that says something changed has to be able to say what.
+
+A group's directory table travelled on `index_sync` alone — a *full* index,
+which the node only ever sends on request. Every ongoing change went out as an
+`index_delta`, which carried files and nothing else. So a root added, removed,
+ejected or plugged by the operator reached every other client's screen only
+when somebody happened to reload the page.
+
+It was hidden by the acks: `root_add_ack` and friends broadcast the new table
+to whoever was connected, so the common cases looked fine. What that could not
+cover is a client connecting mid-change, one whose ack was lost, or — the one
+that surfaced it — the operator's own client, where the ack landed and was then
+overwritten by an index fetched before the node had rebuilt anything.
+
+Additive on the wire (MNP 1.1): a 1.0 client sees a field it does not read.
+"""
+
+from pathlib import Path
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_common.groupbox import PURPOSE_INDEX, unseal
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.transport.wire import index_delta_message, index_sync_message
+
+
+GROUP = "g" * 32
+
+
+def _roots(tmp_path: Path) -> RootSet:
+    for name in ("Films", "Albums"):
+        (tmp_path / name).mkdir()
+    return RootSet.build([
+        {"path": str(tmp_path / "Films"), "writable": True},
+        {"path": str(tmp_path / "Albums"), "removable": True},
+    ])
+
+
+def _index() -> GroupIndex:
+    return GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate(),
+                      gek=generate_gek())
+
+
+def _payload(msg: dict, index: GroupIndex) -> dict:
+    """What a member actually reads, through the seal rather than around it."""
+    return unseal(index.gek, PURPOSE_INDEX, msg["type"], GROUP, msg)
+
+
+class _Delta:
+    base_version = 1
+    version = 2
+    additions: list = []
+    deletions: list = []
+    updates: list = []
+
+
+def test_a_delta_carries_the_directory_table(tmp_path):
+    index = _index()
+    msg = index_delta_message(index, _Delta(), _roots(tmp_path))
+    payload = _payload(msg, index)
+
+    assert [r["name"] for r in payload["roots"]] == ["Films", "Albums"]
+    assert payload["roots"][0]["writable"] is True
+    assert payload["roots"][1]["removable"] is True
+
+
+def test_the_table_is_sealed_with_the_rest(tmp_path):
+    """
+    It is group content, not routing. Only `type`, `v` and `group_id` stay in
+    clear, because a receiver has to route and authenticate before it would
+    trust a decryption.
+    """
+    index = _index()
+    msg = index_delta_message(index, _Delta(), _roots(tmp_path))
+    assert set(msg) - {"type", "v", "group_id"}, "nothing was sealed"
+    assert "roots" not in msg, "the directory table is outside the envelope"
+
+
+def test_a_delta_still_works_without_a_table(tmp_path):
+    """
+    The argument is optional, so an older caller — or a path that has no root
+    set to hand — produces a message a client reads exactly as before.
+    """
+    index = _index()
+    payload = _payload(index_delta_message(index, _Delta()), index)
+    assert "roots" not in payload
+    assert payload["version"] == 2
+
+
+def test_the_table_says_the_same_thing_on_both_messages(tmp_path):
+    """
+    Two encodings of one idea is the drift `wire.py` exists to prevent — it
+    already happened twice, for `index_sync` and for `file_chunk`.
+    """
+    index = _index()
+    roots = _roots(tmp_path)
+    delta = _payload(index_delta_message(index, _Delta(), roots), index)
+    sync = _payload(index_sync_message(index, roots), index)
+    assert delta["roots"] == sync["roots"]
+
+
+def test_the_table_never_carries_a_path(tmp_path):
+    """
+    This message goes to every member. Where a directory lives on the
+    operator's disk is theirs — see test_root_paths_are_operator_only.py.
+    """
+    index = _index()
+    payload = _payload(index_delta_message(index, _Delta(), _roots(tmp_path)),
+                       index)
+    assert not any("path" in r for r in payload["roots"])
+
+
+def test_an_ejected_root_is_visible_in_the_delta(tmp_path):
+    """
+    The case this was written for. An eject changes no file — the entries
+    freeze — so the delta it produces is empty of additions, deletions and
+    updates. Without the table it says literally nothing, which is how a
+    library disappearing from under the group's feet went unannounced.
+    """
+    index = _index()
+    roots = _roots(tmp_path)
+    roots.roots[1].ejected = True
+    roots.roots[1].available = False
+
+    payload = _payload(index_delta_message(index, _Delta(), roots), index)
+    assert payload["additions"] == [] and payload["deletions"] == []
+    assert payload["roots"][1]["ejected"] is True
+    assert payload["roots"][1]["available"] is False
diff --git a/packages/meshbay-node/tests/test_member_upload_policy.py b/packages/meshbay-node/tests/test_member_upload_policy.py
deleted file mode 100644
index b1dc0cb..0000000
--- a/packages/meshbay-node/tests/test_member_upload_policy.py
+++ /dev/null
@@ -1,176 +0,0 @@
-"""
-The operator can close uploading to everyone but themselves.
-
-The point of these tests is the difference between a hidden button and a closed
-door. The interface stops offering the control, which is a courtesy to the
-people who are not trying; **the node refuses the upload**, which is the part
-that holds against someone who is. A member who kept an old tab open, or who
-speaks MNP directly, gets the same answer as everyone else.
-
-Two further things are worth holding:
-
-* the setting is changed by a **signed** operator instruction. A node that took
-  it from an unsigned message would let any member turn it back on, and the
-  control would be a suggestion;
-* it is stored on the **node**, not the hub. A hub that could decide who may
-  write to the operator's disk is a hub with authority over the node, which is
-  the thing this whole design is arranged to avoid.
-"""
-
-import base64
-from pathlib import Path
-
-import pytest
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-
-from meshbay_common.adminop import OP_MEMBER_UPLOAD
-from meshbay_node.indexer.group_index import GroupIndex
-from meshbay_node.roster import Roster
-from meshbay_node.transport.webrtc_server import WebRTCPeerSession
-
-from conftest import one_root
-
-pytestmark = pytest.mark.asyncio
-
-
-def _session(tmp_path: Path, user_id: str, *, member_upload: bool,
-             operator: str | None = None) -> WebRTCPeerSession:
-    shared_root = tmp_path / "shared"
-    shared_root.mkdir(exist_ok=True)
-    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
-    ctx = {
-        "roots": one_root(shared_root),
-        "index": index,
-        "sk_node": index.sk_node,
-        "member_upload": member_upload,
-        "node_user_id": operator,
-    }
-    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
-    session._ctx = ctx
-    session._group_id = None
-    session._user_id = user_id
-    session._pk_user = ""
-    session._uploads = {}
-    session.sent = []
-    session._send = session.sent.append
-    session._audit = lambda *a, **k: None
-    return session
-
-
-def _upload(session, filename="clip.mp4", body=b"bytes"):
-    session._do_file_upload({
-        "filename": filename, "chunk_index": 0, "total_chunks": 1,
-        "data": base64.b64encode(body).decode(),
-    })
-
-
-def _uploads_dir(session) -> Path:
-    return session._ctx["roots"].upload_root.path / "uploads"
-
-
-# ── The door, not the button ────────────────────────────────────────────────
-
-async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path):
-    session = _session(tmp_path, "member-1", member_upload=False,
-                       operator="the-operator")
-    _upload(session)
-
-    assert not (_uploads_dir(session) / "clip.mp4").exists(), (
-        "the file was written even though uploading is off — the setting is "
-        "decorative and the hidden button was the whole control")
-    refusal = [m for m in session.sent if m.get("type") == "error"]
-    assert refusal and refusal[0].get("code") == "member_upload_off"
-
-
-async def test_the_operator_can_still_upload(tmp_path):
-    """Otherwise turning it off locks the operator out of their own node, and
-    the only way back is a config file and a restart."""
-    session = _session(tmp_path, "the-operator", member_upload=False,
-                       operator="the-operator")
-    _upload(session)
-
-    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
-
-
-async def test_members_upload_normally_when_it_is_on(tmp_path):
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    _upload(session)
-
-    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
-
-
-async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path):
-    """An existing node's context has no such key. The absence must read as
-    "allowed", or upgrading the node silently closes every group."""
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    del session._ctx["member_upload"]
-    _upload(session)
-
-    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
-
-
-# ── Who may change it ───────────────────────────────────────────────────────
-
-async def test_changing_it_needs_a_signature(tmp_path):
-    """
-    The request only ever produces a challenge. Nothing is applied until a
-    signature over the transcript verifies — the same path as removing a member.
-    """
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    session._has_admin_authority = lambda: True
-    issued = []
-    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
-
-    session._do_member_upload({"allowed": False})
-
-    assert issued == [(OP_MEMBER_UPLOAD, "off")]
-    assert session._ctx["member_upload"] is True, "applied before it was signed"
-
-
-async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
-    """The operator is shown the subject before signing. "member_upload" tells
-    them nothing; "off" tells them what they are about to do."""
-    session = _session(tmp_path, "op", member_upload=False, operator="op")
-    session._has_admin_authority = lambda: True
-    issued = []
-    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
-
-    session._do_member_upload({"allowed": True})
-
-    assert issued == [(OP_MEMBER_UPLOAD, "on")]
-
-
-async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
-    session = _session(tmp_path, "member-1", member_upload=True,
-                       operator="the-operator")
-    session._has_admin_authority = lambda: False
-
-    session._do_member_upload({"allowed": False})
-
-    assert [m for m in session.sent if m.get("type") == "error"]
-
-
-# ── Where it is stored ──────────────────────────────────────────────────────
-
-async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
-    roster = Roster(db_path=tmp_path / "roster.db")
-    await roster.open()
-    try:
-        assert await roster.member_upload_allowed("g1") is True, (
-            "absent must mean allowed, or an upgrade closes every group")
-        await roster.set_member_upload("g1", False, set_by="op")
-        assert await roster.member_upload_allowed("g1") is False
-    finally:
-        await roster.close()
-
-    reopened = Roster(db_path=tmp_path / "roster.db")
-    await reopened.open()
-    try:
-        assert await reopened.member_upload_allowed("g1") is False
-        assert await reopened.member_upload_allowed("g2") is True, (
-            "one group's setting must not answer for another")
-    finally:
-        await reopened.close()
diff --git a/packages/meshbay-node/tests/test_node_status.py b/packages/meshbay-node/tests/test_node_status.py
index b56eb6e..091b1db 100644
--- a/packages/meshbay-node/tests/test_node_status.py
+++ b/packages/meshbay-node/tests/test_node_status.py
@@ -255,7 +255,7 @@ async def test_add_root_creates_directory_and_returns_info(tmp_path):
 
     from meshbay_node.config import NodeConfig, GroupConfig, RootSpec
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(shared), name="shared", kind="generic", upload=True),
+        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
     ])
 
     conf = tmp_path / "node.toml"
@@ -295,7 +295,7 @@ async def test_remove_root_requires_at_least_one_remaining(tmp_path):
 
     from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(shared), name="shared", kind="generic", upload=True),
+        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
     ])
     node_cfg = NodeConfig.__new__(NodeConfig)
     node_cfg.groups = [cfg]
@@ -314,16 +314,22 @@ async def test_remove_root_requires_at_least_one_remaining(tmp_path):
         await ops.remove_root(state, GROUP, "shared")
 
 
-async def test_remove_root_refuses_upload_root(tmp_path):
-    d1 = tmp_path / "uploads"
+async def test_removing_a_writable_root_is_allowed(tmp_path):
+    """
+    It used to be refused: with one designated upload root, removing it left
+    the group with nowhere to put an upload and no way to say so. Several roots
+    can be writable now, and a group with none is a valid read-only group — so
+    the refusal would be protecting a state that is no longer special.
+    """
+    d1 = tmp_path / "incoming"
     d2 = tmp_path / "shared"
     d1.mkdir()
     d2.mkdir()
 
     from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(d1), name="uploads", kind="generic", upload=True),
-        RootSpec(path=str(d2), name="shared", kind="generic", upload=False),
+        RootSpec(path=str(d1), name="incoming", kind="generic", writable=True),
+        RootSpec(path=str(d2), name="shared", kind="generic", writable=False),
     ])
     node_cfg = NodeConfig.__new__(NodeConfig)
     node_cfg.groups = [cfg]
@@ -331,7 +337,7 @@ async def test_remove_root_refuses_upload_root(tmp_path):
     conf = tmp_path / "node.toml"
     conf.write_text(
         f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
-        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "uploads"\n  upload = true\n\n'
+        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "incoming"\n  writable = true\n\n'
         f'  [[groups.roots]]\n  path = "{d2}"\n  name = "shared"\n')
     roots = RootSet.build([asdict(r) for r in cfg.roots])
     index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
@@ -340,8 +346,97 @@ async def test_remove_root_refuses_upload_root(tmp_path):
         "config_path": str(conf),
         "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
     }
-    with pytest.raises(ops.OpError, match="upload root"):
-        await ops.remove_root(state, GROUP, "uploads")
+    result = await ops.remove_root(state, GROUP, "incoming")
+    assert result["status"] == "removed"
+    assert [r["name"] for r in result["roots"]] == ["shared"]
+    assert conf.read_text().count("[[groups.roots]]") == 1
+
+
+async def test_update_root_rewrites_the_flags_in_node_toml(tmp_path):
+    """
+    The flags live in the operator's config file, so they survive a restart —
+    and the file is hand-written and full of comments, so the change is a line
+    edit rather than a round trip through a TOML writer that would discard
+    every one of them.
+    """
+    d1 = tmp_path / "media"
+    d1.mkdir()
+
+    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
+    cfg = GroupConfig(id=GROUP, name="test", roots=[
+        RootSpec(path=str(d1), name="media", kind="generic", writable=False),
+    ])
+    node_cfg = NodeConfig.__new__(NodeConfig)
+    node_cfg.groups = [cfg]
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
+        f'  [[groups.roots]]\n'
+        f'  # the operator explained this one to themselves\n'
+        f'  path = "{d1}"\n  name = "media"\n')
+    roots = RootSet.build([asdict(r) for r in cfg.roots])
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    state = {
+        "config": node_cfg,
+        "config_path": str(conf),
+        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
+    }
+
+    result = await ops.update_root(state, GROUP, "media",
+                                   writable=True, removable=True)
+    assert result["status"] == "updated"
+    text = conf.read_text()
+    assert "writable = true" in text
+    assert "removable = true" in text
+    assert "the operator explained this one to themselves" in text, (
+        "the config file was rewritten instead of edited")
+
+    # And the live root set agrees immediately, without waiting for a reload:
+    # the loopback API reads it, and an operator who toggles a switch and sees
+    # it snap back assumes the change did not take.
+    assert roots.roots[0].writable is True
+    assert roots.roots[0].removable is True
+
+    # A second call that changes nothing must not append a duplicate line.
+    await ops.update_root(state, GROUP, "media", writable=True, removable=True)
+    assert conf.read_text().count("writable =") == 1
+
+
+async def test_update_root_replaces_a_legacy_upload_line(tmp_path):
+    """
+    A config written before the refactor says `upload = true`. Leaving it in
+    place next to a new `writable` line would give the file two answers, and
+    `RootSet.build` prefers `writable` — so the stale one would sit there
+    contradicting the running node for as long as anyone read it.
+    """
+    d1 = tmp_path / "media"
+    d1.mkdir()
+
+    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
+    cfg = GroupConfig(id=GROUP, name="test", roots=[
+        RootSpec(path=str(d1), name="media", kind="generic", writable=True),
+    ])
+    node_cfg = NodeConfig.__new__(NodeConfig)
+    node_cfg.groups = [cfg]
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
+        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "media"\n'
+        f'  upload = true\n')
+    roots = RootSet.build([asdict(r) for r in cfg.roots])
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    state = {
+        "config": node_cfg,
+        "config_path": str(conf),
+        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
+    }
+
+    await ops.update_root(state, GROUP, "media", writable=False)
+    text = conf.read_text()
+    assert "upload = true" not in text
+    assert "writable = false" in text
 
 
 async def test_remove_root_succeeds_with_two_roots(tmp_path):
@@ -352,8 +447,8 @@ async def test_remove_root_succeeds_with_two_roots(tmp_path):
 
     from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
     cfg = GroupConfig(id=GROUP, name="test", roots=[
-        RootSpec(path=str(d1), name="dir1", kind="generic", upload=True),
-        RootSpec(path=str(d2), name="dir2", kind="generic", upload=False),
+        RootSpec(path=str(d1), name="dir1", kind="generic", writable=True),
+        RootSpec(path=str(d2), name="dir2", kind="generic", writable=False),
     ])
     node_cfg = NodeConfig.__new__(NodeConfig)
     node_cfg.groups = [cfg]
diff --git a/packages/meshbay-node/tests/test_ops.py b/packages/meshbay-node/tests/test_ops.py
index 92e32bf..c118b5a 100644
--- a/packages/meshbay-node/tests/test_ops.py
+++ b/packages/meshbay-node/tests/test_ops.py
@@ -11,12 +11,14 @@ call them.
 
 import asyncio
 import inspect
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
+from types import SimpleNamespace
 
 import pytest
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from meshbay_node import ops
 from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
 from meshbay_node.transport.quic_server import Denylist
 
 from conftest import one_root
@@ -64,8 +66,8 @@ def test_the_http_adapter_adds_no_logic():
     # 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",
-                     "add_root", "remove_root", "set_member_upload",
-                     "reload_config"):
+                     "add_root", "remove_root", "update_root",
+                     "eject_root", "plug_root", "reload_config"):
         start = source.index(f"async def {endpoint}(")
         body = source[start:start + 700]
         assert "_op(" in body.split("\n\n")[0] + body, (
@@ -181,25 +183,103 @@ async def test_an_unhosted_group_offers_what_it_does_host(tmp_path):
     assert exc.value.extra.get("available")
 
 
-# ── Upload policy (set_member_upload) ───────────────────────────────────────
+# ── Upload policy (per-root writable) ───────────────────────────────────────
 
-async def test_set_member_upload_toggles_and_persists(tmp_path):
+async def test_the_group_wide_upload_switch_is_gone(tmp_path):
+    """
+    `set_member_upload` was the whole of the old policy, and it is deliberately
+    not here any more — RO/RW on the root replaced it. A wrapper kept "for
+    compatibility" would be a second way to decide who writes to the operator's
+    disk, and two answers to that question is how C1 and C6 both happened.
+    """
+    assert not hasattr(ops, "set_member_upload")
+    from meshbay_node.roster import Roster
+    assert not hasattr(Roster, "set_member_upload")
+    assert not hasattr(Roster, "member_upload_allowed")
+
+
+async def test_eject_and_plug_persist_through_the_roster(tmp_path):
+    """
+    The state has to outlive the process: an operator ejects a drive, unplugs
+    it, and restarts the node — and the rescan that follows must not read the
+    empty mount point as an erased library.
+    """
     from meshbay_node.roster import Roster
     state = _state(tmp_path)
+    usb = tmp_path / "USB"
+    usb.mkdir()
+    state["groups_ctx"]["g" * 32]["roots"] = RootSet.build(
+        [{"path": str(usb), "removable": True, "writable": True}])
+    state["config"] = SimpleNamespace(
+        groups=[SimpleNamespace(id="g" * 32, roots=[])])
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     state["roster"] = roster
     state["node_user_id"] = "operator"
+    try:
+        out = await ops.eject_root(state, "g" * 32, "USB")
+        assert out["status"] == "ejected"
+        assert await roster.ejected_roots("g" * 32) == {"usb"}
+        assert out["roots"][0]["ejected"] is True
+        assert out["roots"][0]["available"] is False
 
-    out = await ops.set_member_upload(state, "g" * 32, True)
+        out = await ops.plug_root(state, "g" * 32, "USB")
+        assert out["status"] == "plugged"
+        assert await roster.ejected_roots("g" * 32) == set()
+    finally:
+        await roster.close()
 
-    assert out["allowed"] is True
-    assert state["groups_ctx"]["g" * 32]["member_upload"] is True
 
-    out2 = await ops.set_member_upload(state, "g" * 32, False)
+async def test_a_root_that_is_not_removable_cannot_be_ejected(tmp_path):
+    """
+    Eject means "I am about to unplug this". On a directory that is not on a
+    removable device it would hide a library with no way for the safety net to
+    notice anything happened, and nothing to plug back in.
+    """
+    from meshbay_node.roster import Roster
+    state = _state(tmp_path)
+    fixed = tmp_path / "Fixed"
+    fixed.mkdir()
+    state["groups_ctx"]["g" * 32]["roots"] = RootSet.build(
+        [{"path": str(fixed), "writable": True}])
+    state["config"] = SimpleNamespace(
+        groups=[SimpleNamespace(id="g" * 32, roots=[])])
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state["roster"] = roster
+    try:
+        with pytest.raises(ops.OpError, match="removable"):
+            await ops.eject_root(state, "g" * 32, "Fixed")
+    finally:
+        await roster.close()
 
-    assert out2["allowed"] is False
-    assert state["groups_ctx"]["g" * 32]["member_upload"] is False
+
+async def test_plugging_a_drive_that_is_not_there_is_refused(tmp_path):
+    """
+    Clearing the flag while the device is still absent would restart the
+    watchdog on a missing path and hand the next reconcile an empty directory —
+    the deletion storm the eject was there to prevent, produced by the recovery.
+    """
+    from meshbay_node.roster import Roster
+    state = _state(tmp_path)
+    usb = tmp_path / "USB"
+    usb.mkdir()
+    roots = RootSet.build([{"path": str(usb), "removable": True}])
+    roots.roots[0].ejected = True
+    roots.roots[0].available = False
+    state["groups_ctx"]["g" * 32]["roots"] = roots
+    state["config"] = SimpleNamespace(
+        groups=[SimpleNamespace(id="g" * 32, roots=[])])
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state["roster"] = roster
+    usb.rmdir()
+    try:
+        with pytest.raises(ops.OpError, match="device connected"):
+            await ops.plug_root(state, "g" * 32, "USB")
+        assert roots.roots[0].ejected is True
+    finally:
+        await roster.close()
 
 
 # ── Reload ──────────────────────────────────────────────────────────────────
@@ -279,13 +359,47 @@ def test_update_node_toml_forces_lf_and_keeps_standalone_comments(tmp_path):
 
 
 def test_a_backslash_path_written_into_node_toml_stays_parseable():
-    # attach_group / add_root / init embed a directory into a TOML basic string.
-    # A raw Windows path there (drive + backslash + "Users" + ...) is a parse
-    # error since backslash sequences are escapes; the code writes as_posix().
+    """
+    attach_group and add_root embed a directory into a TOML basic string. A raw
+    Windows path there is a parse error, because backslash sequences are escapes
+    (`\\U`, `\\a`, ...); the code writes `as_posix()` and pathlib reads `/` back
+    on Windows.
+
+    `PureWindowsPath`, not `Path`: on this suite's usual machine `Path` is a
+    `PosixPath`, where a backslash is an ordinary filename character and
+    `as_posix()` converts nothing — so the test modelled the wrong platform and
+    failed everywhere except the one it was written for. Naming the flavour
+    explicitly is what makes it the same assertion on all three.
+    """
     import tomllib
     bs = chr(92)
     win_dir = f"C:{bs}Users{bs}alice{bs}Media"
-    assert tomllib.loads(f'path = "{Path(win_dir).as_posix()}"\n')["path"] == \
-        "C:/Users/alice/Media"
+
+    assert tomllib.loads(
+        f'path = "{PureWindowsPath(win_dir).as_posix()}"\n'
+    )["path"] == "C:/Users/alice/Media"
+
     with pytest.raises(tomllib.TOMLDecodeError):
         tomllib.loads(f'path = "{win_dir}"\n')   # the bug this guards against
+
+
+def test_every_path_written_into_node_toml_goes_through_as_posix():
+    """
+    The half the round trip above cannot see.
+
+    Proving `as_posix()` produces a parseable string says nothing about whether
+    the code calls it, and this is a defect no Linux machine can reproduce: the
+    config is written, parsed and served correctly here, and fails on the
+    operator's Windows box. So the source is read for the shape instead —
+    weak evidence, and the only kind available for a platform the suite does
+    not run on.
+    """
+    import re
+    source = inspect.getsource(ops)
+    # Every f-string interpolation that lands on the right of a TOML `path =`.
+    writes = re.findall(r'path\s*=\s*\\?"\{([^}]+)\}', source)
+    assert writes, "no TOML path writer found — did the config writer move?"
+    for expr in writes:
+        assert "as_posix()" in expr, (
+            f'node.toml path written as `{expr}` — a Windows path needs '
+            f'as_posix(), or the file it lands in will not parse')
diff --git a/packages/meshbay-node/tests/test_rename_reenrichment.py b/packages/meshbay-node/tests/test_rename_reenrichment.py
index 7a77368..f6761d9 100644
--- a/packages/meshbay-node/tests/test_rename_reenrichment.py
+++ b/packages/meshbay-node/tests/test_rename_reenrichment.py
@@ -37,8 +37,8 @@ def _free_port() -> int:
 
 
 class _StubRoster:
-    async def video_root(self, group_id):
-        return "shared"
+    async def app_directories(self, group_id, app_key):
+        return ["shared"] if app_key == "video" else []
 
 
 class _SpyEnricher:
diff --git a/packages/meshbay-node/tests/test_replug_restores_enrichment.py b/packages/meshbay-node/tests/test_replug_restores_enrichment.py
new file mode 100644
index 0000000..04a06ae
--- /dev/null
+++ b/packages/meshbay-node/tests/test_replug_restores_enrichment.py
@@ -0,0 +1,318 @@
+"""
+A root that comes back keeps its Videos/Music/Photos metadata.
+
+Reported live: a removable root ejected from the Files app and plugged back
+in returned with its files and without its albums. Music showed "no music
+found", and it did not come back.
+
+`plug_root` has to re-walk the root — the drive may have changed while it
+was away — and `_scan_root` produces bare entries: `_hash_or_cached` fills
+id/name/path/size/type and nothing else. Every enrichment field went with
+the old object, and the Music tag fields are cached nowhere by design
+(`enrich_audio.py` re-reads them so a rename can re-derive the filename
+fallback).
+
+Two gates then stopped anything from filling them in again:
+
+* enrichment is scheduled for `delta.additions`, and ejecting broadcasts
+  nothing — the last snapshot still held those ids, so the rebuilt entries
+  diffed as *updates*;
+* `_enrich_new_*_entries` skips anything in `_enriched_attempted`, which is
+  only discarded for `delta.deletions` — and dropping and rescanning inside
+  one call broadcasts no deletion either.
+
+Only a restart cleared both, an empty snapshot making every entry an
+addition. That is why it looked like it might fix itself and never did.
+
+Re-enriching is now the *fallback*, not the fix. An entry's id is its
+content hash, so one that comes back under the same id, name and path is
+the same bytes in the same place and its enrichment still holds:
+`_rescan_root` carries those fields across. Re-deriving them instead meant
+tag reads, ffprobe runs and rate-limited lookups — measured at 14 seconds
+of empty Music tab on a real library with a cold cache, which to the
+operator is indistinguishable from the original bug.
+
+What these assert is therefore the field on the entry, not a call to an
+enricher. Counting calls is what made an earlier version of this file pass
+while the operator still watched their albums vanish.
+
+The same drop-and-rescan runs in `reconcile()` — "Root %r is back" — so a
+USB drive that falls off and returns on its own hits all of this without
+anybody touching the UI.
+"""
+
+import asyncio
+import os
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.crypto import generate_gek
+from meshbay_node.config import (Config, GroupConfig, HubConfig, KeystoreConfig,
+                                 NodeConfig)
+from meshbay_node.daemon import NodeDaemon
+from meshbay_node.indexer import DirectoryIndexer
+from meshbay_node.indexer.enrich_audio import AudioEnricher
+from meshbay_node.media_cache import MediaCache
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+# Above indexer.py's MIN_AUDIO_SIZE_BYTES, or nothing would be indexed.
+_AUDIO_BYTES = os.urandom(60 * 1024)
+
+
+def _free_port() -> int:
+    import socket
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+class _CountingEnricher:
+    """Stands in for an enricher: records who it was asked to enrich."""
+
+    def __init__(self):
+        self.spawned: list[str] = []
+
+    def spawn(self, entry, file_path, on_done, boundary=None):
+        self.spawned.append(entry.name)
+
+
+async def _daemon(tmp_path, shared, group_id):
+    config = Config(
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
+        node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
+        groups=[GroupConfig(
+            id=group_id, name="test-group", shared_dir=str(shared),
+            visibility="private", quic_port=_free_port(),
+        )],
+        keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
+        data_dir=tmp_path / "data",
+    )
+    daemon = NodeDaemon(config)
+    daemon._broadcast_coalesce_secs = 0.01
+    daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
+    await daemon._media_cache.open()
+    daemon._roster = Roster(db_path=tmp_path / "roster.db")
+    await daemon._roster.open()
+    return daemon
+
+
+async def _settled(daemon, indexer):
+    await daemon._on_index_change(indexer)
+    await asyncio.sleep(0.05)
+
+
+async def _library(tmp_path, group_id, *, enricher=None):
+    """A one-track library under <root>/<artist>/<album>, enriched once."""
+    library = tmp_path / "music"
+    (library / "an artist" / "a record").mkdir(parents=True)
+    (library / "an artist" / "a record" / "01 first track.mp3").write_bytes(_AUDIO_BYTES)
+
+    daemon = await _daemon(tmp_path, library, group_id)
+    daemon._audio_enricher = enricher or AudioEnricher(daemon._media_cache)
+    await daemon._roster.set_app_directories(
+        group_id, "music", ["music"], set_by="op")
+
+    roots = RootSet.build([{"path": str(library), "removable": True}])
+    indexer = DirectoryIndexer(
+        roots=roots, group_id=group_id,
+        sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+    await indexer.initial_scan()
+    await _settled(daemon, indexer)
+    await asyncio.sleep(0.4)  # the enricher runs off the broadcast
+    return daemon, indexer, roots, library
+
+
+def _only(indexer):
+    return next(iter(indexer.index.entries))
+
+
+# ── The operator's own eject and plug ────────────────────────────────────────
+
+async def test_the_albums_are_still_there_after_a_replug(tmp_path):
+    """
+    No tags are written: `enrich_audio._artist_album_from_ancestors` derives
+    artist and album from the folder names when a file has none, which is the
+    <library>/<artist>/<album>/<track> layout this was reported against.
+    """
+    group_id = "a" * 32
+    daemon, indexer, _, _ = await _library(tmp_path, group_id)
+    try:
+        before = _only(indexer)
+        assert before.album == "a record" and before.artist == "an artist", (
+            f"the first pass never filled the fields: {before}")
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        after = _only(indexer)
+        assert after.album == "a record" and after.artist == "an artist", (
+            "the entry came back from the rescan with no album — this is what "
+            "an empty Music tab after a replug looks like on the node")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+async def test_the_fields_survive_without_re_deriving_them(tmp_path):
+    """
+    Carried across, not recomputed. Re-deriving is correct and far too slow:
+    on a real library with a cold metadata cache it left Music empty for 14
+    seconds, and an operator who looks in that window sees the bug.
+    """
+    group_id = "b" * 32
+    enricher = _CountingEnricher()
+    daemon, indexer, _, _ = await _library(tmp_path, group_id, enricher=enricher)
+    try:
+        assert enricher.spawned == ["01 first track.mp3"]
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        assert enricher.spawned == ["01 first track.mp3"], (
+            "an unchanged file was enriched a second time — the whole point "
+            "of the content hash is that it did not need to be")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+async def test_who_uploaded_a_file_survives_it_too(tmp_path):
+    """
+    `uploader_id`/`uploader_pk` are the same shape of field — set once, on an
+    entry, readable from nowhere on disk — and they decide who may delete the
+    file. Losing them to a replug quietly takes a right away.
+    """
+    group_id = "c" * 32
+    daemon, indexer, _, _ = await _library(tmp_path, group_id)
+    try:
+        entry = _only(indexer)
+        entry.uploader_id = "alice"
+        entry.uploader_pk = "a-pinned-key"
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        after = _only(indexer)
+        assert after.uploader_id == "alice" and after.uploader_pk == "a-pinned-key"
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+# ── A drive that leaves and returns on its own ──────────────────────────────
+
+async def test_a_root_that_returns_on_its_own_is_treated_the_same(tmp_path):
+    """
+    `reconcile()` rescans a root that reappears without anyone asking — a USB
+    drive re-mounting. Same drop-and-rescan, so it lost the same fields, with
+    no click anywhere to blame it on.
+    """
+    group_id = "d" * 32
+    daemon, indexer, roots, _ = await _library(tmp_path, group_id)
+    try:
+        assert _only(indexer).album == "a record"
+
+        roots.roots[0].available = False
+        await indexer.reconcile()
+        await _settled(daemon, indexer)
+        await indexer.reconcile()
+        await _settled(daemon, indexer)
+        await asyncio.sleep(0.4)
+
+        assert _only(indexer).album == "a record", (
+            "a drive that fell off and came back left the library with no "
+            "metadata")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+# ── What genuinely does have to be re-derived ───────────────────────────────
+
+async def test_a_track_moved_while_the_drive_was_away_is_enriched_again(tmp_path):
+    """
+    The counter-case, and the reason the carry-over is keyed on name and path
+    as well as id. `artist`, `album`, `display_title` and `track_no` all fall
+    back to the folder and filename when a file carries no tags, so the same
+    bytes under a new name are not the same metadata. Those are the entries
+    the daemon still re-enriches, off `rescanned_ids`.
+    """
+    group_id = "e" * 32
+    enricher = _CountingEnricher()
+    daemon, indexer, _, library = await _library(
+        tmp_path, group_id, enricher=enricher)
+    try:
+        assert enricher.spawned == ["01 first track.mp3"]
+
+        moved = library / "another artist" / "another record"
+        moved.mkdir(parents=True)
+        (library / "an artist" / "a record" / "01 first track.mp3").rename(
+            moved / "01 first track.mp3")
+
+        indexer.eject_root("music")
+        await indexer.plug_root("music")
+        await _settled(daemon, indexer)
+
+        assert enricher.spawned == ["01 first track.mp3"] * 2, (
+            "the file is under a different artist and album now; carrying the "
+            "old ones across would file it under a folder it left")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
+
+
+async def test_videos_and_photos_are_covered_by_the_same_path(tmp_path):
+    """
+    Nothing here is specific to Music — Videos and Photos lost their durations,
+    titles and thumbnails the same way. Music is simply where it shows up
+    loudest: a track with no tags has no album to file it under, so the app
+    goes empty rather than merely plain.
+    """
+    group_id = "f" * 32
+    library = tmp_path / "media"
+    (library / "films").mkdir(parents=True)
+    (library / "films" / "clip.mkv").write_bytes(os.urandom(60 * 1024))
+    (library / "album").mkdir(parents=True)
+    (library / "album" / "shot.jpg").write_bytes(os.urandom(60 * 1024))
+
+    daemon = await _daemon(tmp_path, library, group_id)
+    video, photo = _CountingEnricher(), _CountingEnricher()
+    daemon._enricher, daemon._photo_enricher = video, photo
+    try:
+        await daemon._roster.set_app_directories(
+            group_id, "video", ["media/films"], set_by="op")
+        await daemon._roster.set_app_directories(
+            group_id, "photo", ["media/album"], set_by="op")
+
+        roots = RootSet.build([{"path": str(library), "removable": True}])
+        indexer = DirectoryIndexer(
+            roots=roots, group_id=group_id,
+            sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
+        await indexer.initial_scan()
+        await _settled(daemon, indexer)
+        assert video.spawned == ["clip.mkv"] and photo.spawned == ["shot.jpg"]
+
+        by_name = {e.name: e for e in indexer.index.entries}
+        by_name["clip.mkv"].duration = 1234
+        by_name["shot.jpg"].thumb_hash = "a-thumbnail"
+
+        indexer.eject_root("media")
+        await indexer.plug_root("media")
+        await _settled(daemon, indexer)
+
+        back = {e.name: e for e in indexer.index.entries}
+        assert back["clip.mkv"].duration == 1234, "the film lost its probe"
+        assert back["shot.jpg"].thumb_hash == "a-thumbnail", (
+            "the photo lost its thumbnail")
+        assert video.spawned == ["clip.mkv"] and photo.spawned == ["shot.jpg"], (
+            "unchanged files were probed and thumbnailed all over again")
+    finally:
+        await daemon._media_cache.close()
+        await daemon._roster.close()
diff --git a/packages/meshbay-node/tests/test_root_availability.py b/packages/meshbay-node/tests/test_root_availability.py
index 0201c1f..d514dee 100644
--- a/packages/meshbay-node/tests/test_root_availability.py
+++ b/packages/meshbay-node/tests/test_root_availability.py
@@ -26,9 +26,12 @@ from meshbay_node.roots import RootSet
 pytestmark = pytest.mark.asyncio
 
 
-def _roots(*paths: Path) -> RootSet:
+def _roots(*paths: Path, removable: bool = False) -> RootSet:
     specs = [{"path": str(p)} for p in paths]
-    specs[0]["upload"] = True
+    specs[0]["writable"] = True
+    if removable:
+        for spec in specs:
+            spec["removable"] = True
     return RootSet.build(specs)
 
 
@@ -117,7 +120,9 @@ async def test_members_are_told_which_roots_are_unavailable(tmp_path):
 
     idx = await _indexer(_roots(films))
     assert idx.index.roots == [
-        {"name": "Films", "kind": "generic", "available": True, "upload": True}]
+        {"name": "Films", "kind": "generic", "available": True,
+         "writable": True, "removable": False, "ejected": False,
+         "upload": True}]
 
     (films / "a.mkv").unlink()
     films.rmdir()
diff --git a/packages/meshbay-node/tests/test_root_eject.py b/packages/meshbay-node/tests/test_root_eject.py
new file mode 100644
index 0000000..f57fb92
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_eject.py
@@ -0,0 +1,267 @@
+"""
+Safe eject, and the surprise unplug it exists to survive.
+
+`test_root_availability.py` pins the freeze: a root that goes away keeps its
+entries. This pins the half the operator drives — telling the node the drive is
+about to leave, and telling it the drive is back.
+
+The distinction that makes any of this work is that `ejected` and `is_live()`
+are separate answers. Between clicking Eject and physically unplugging, the
+directory is still readable; a design that recomputed availability from the
+filesystem alone would flip the root straight back to available and start
+serving files from a disk somebody has their hand on.
+
+The other property here is that the flag is *persisted*. It reached the roster
+in the first implementation and was never read back, so a restart — which is
+exactly what an operator does after noticing a drive fell off — silently undid
+the eject, and the next scan read an empty mount point as an erased library.
+"""
+
+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
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+
+def _roots(*paths: Path, removable: bool = True) -> RootSet:
+    return RootSet.build([
+        {"path": str(p), "removable": removable} for p in paths])
+
+
+async def _indexer(roots: RootSet, **kw) -> DirectoryIndexer:
+    idx = DirectoryIndexer(roots=roots, group_id="g" * 32,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None, **kw)
+    await idx.initial_scan()
+    return idx
+
+
+def _names(idx: DirectoryIndexer) -> set[str]:
+    return {e.name for e in idx.index.entries}
+
+
+# ── The two states are not the same question ─────────────────────────────────
+
+async def test_ejecting_hides_a_root_that_is_still_readable(tmp_path):
+    """
+    The whole point of an eject button: the operator says the drive is leaving
+    *before* it leaves. The directory is still there and still readable at this
+    moment, so anything deriving availability from the filesystem would refuse
+    to believe it.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+
+    assert films.is_dir(), "the drive has not been unplugged yet"
+    assert roots.roots[0].is_live() is True
+    assert roots.roots[0].available is False
+    assert idx.index.roots[0]["ejected"] is True
+    assert idx.index.roots[0]["available"] is False
+
+
+async def test_an_eject_freezes_entries_rather_than_dropping_them(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))
+    idx.eject_root("Films")
+
+    assert _names(idx) == {"a.mkv", "b.mkv"}, "eject deleted entries"
+
+
+async def test_reconciling_does_not_un_eject_a_root(tmp_path):
+    """
+    The backstop runs every minute regardless. An ejected root whose directory
+    is still readable must stay ejected, or the operator's eject lasts until
+    the next tick.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+    await idx.reconcile()
+
+    assert roots.roots[0].ejected is True
+    assert roots.roots[0].available is False
+
+
+async def test_plugging_back_relists_the_files(tmp_path):
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+    await idx.plug_root("Films")
+
+    assert roots.roots[0].ejected is False
+    assert roots.roots[0].available is True
+    assert _names(idx) == {"a.mkv"}
+
+
+async def test_what_changed_while_unplugged_is_picked_up_on_plug(tmp_path):
+    """
+    A drive people take away comes back different. The plug pass has to see
+    that, or the index describes a library that no longer exists on the disk
+    the node is about to serve from.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+    idx.eject_root("Films")
+
+    (films / "a.mkv").unlink()
+    (films / "c.mkv").write_bytes(b"c")
+
+    await idx.plug_root("Films")
+    assert _names(idx) == {"c.mkv"}
+
+
+# ── The surprise unplug ──────────────────────────────────────────────────────
+
+async def test_a_removable_root_that_vanishes_is_auto_ejected(tmp_path):
+    """
+    Nobody clicks Eject when they are in a hurry. A removable root whose path
+    disappears is treated as ejected rather than merely unavailable, so it does
+    not silently come back the moment the same mount point is readable again —
+    which on a machine with automount is any other drive, or an empty stub.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films)
+    idx = await _indexer(roots)
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+
+    assert roots.roots[0].ejected is True
+    assert _names(idx) == {"a.mkv"}, "the library was treated as erased"
+
+
+async def test_a_non_removable_root_is_not_auto_ejected(tmp_path):
+    """
+    The counter-property. Auto-eject requires the operator to have said the
+    device is removable; an ordinary directory that briefly fails to stat must
+    keep the old behaviour and come back on its own.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    roots = _roots(films, removable=False)
+    idx = await _indexer(roots)
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+    assert roots.roots[0].ejected is False
+    assert roots.roots[0].available is False
+
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+    await idx.reconcile()
+    assert roots.roots[0].available is True
+
+
+async def test_an_auto_eject_is_reported_so_it_can_be_persisted(tmp_path):
+    """
+    The flag has to outlive the process. The first version of this set it in
+    memory only, so restarting the node — which is what an operator does after
+    noticing a drive fell off — cleared it, and the scan that followed read the
+    empty mount point as a deletion of the whole library.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    (films / "a.mkv").write_bytes(b"a")
+
+    seen: list[tuple[str, bool]] = []
+
+    async def record(name: str, ejected: bool) -> None:
+        seen.append((name, ejected))
+
+    roots = _roots(films)
+    idx = await _indexer(roots, on_root_ejected=record)
+
+    (films / "a.mkv").unlink()
+    films.rmdir()
+    await idx.reconcile()
+
+    assert seen == [("Films", True)]
+
+    # And only once, however many times the backstop runs afterwards.
+    await idx.reconcile()
+    await idx.reconcile()
+    assert seen == [("Films", True)]
+
+
+# ── Restoring the flag ───────────────────────────────────────────────────────
+
+async def test_a_root_built_as_ejected_starts_unavailable(tmp_path):
+    """
+    What the daemon does with what the roster remembers. `available` must not
+    be left at its default `True` here, or the group serves a drive that is not
+    there for as long as it takes the first reconcile to run.
+    """
+    films = tmp_path / "Films"
+    films.mkdir()
+    roots = RootSet.build([{"path": str(films), "removable": True,
+                            "ejected": True}])
+    assert roots.roots[0].ejected is True
+    assert roots.roots[0].available is False
+
+
+async def test_the_roster_round_trips_the_ejected_set(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.ejected_roots("g1") == set()
+
+        await roster.set_root_ejected("g1", "Films", True, set_by="op")
+        await roster.set_root_ejected("g1", "Music", False, set_by="op")
+        assert await roster.ejected_roots("g1") == {"films"}
+
+        # Another group's drives are its own.
+        assert await roster.ejected_roots("g2") == set()
+
+        await roster.set_root_ejected("g1", "Films", False, set_by="op")
+        assert await roster.ejected_roots("g1") == set()
+    finally:
+        await roster.close()
+
+
+async def test_the_ejected_key_is_case_folded(tmp_path):
+    """
+    Root names are compared without regard to case everywhere else, and a key
+    that did not fold would let `Films` and `films` disagree about the same
+    drive — on Windows and macOS, the same directory.
+    """
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        await roster.set_root_ejected("g1", "FILMS", True, set_by="op")
+        assert await roster.ejected_roots("g1") == {"films"}
+        assert Roster.root_ejected_key("Films") == Roster.root_ejected_key("FILMS")
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py b/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py
new file mode 100644
index 0000000..976af82
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_ops_reach_the_live_set.py
@@ -0,0 +1,361 @@
+"""
+Adding or removing a root has to reach the running node, not only node.toml.
+
+Two front doors do this — the loopback API and a signed MNP op — and `ops.py`
+exists so they behave identically. They did not. The loopback path fired the
+daemon's `reload_fn`, which re-reads node.toml and builds a fresh `RootSet`;
+the MNP path instead re-pointed the indexer at `groups_ctx[gid]["roots"]`, the
+very object the op had just been asked about. `DirectoryIndexer.retarget`
+decides what to scan by diffing the names it holds against the ones it is
+given, so a set compared against itself scans nothing and drops nothing.
+
+A directory added from a browser therefore reached node.toml and was invisible
+everywhere else until a restart — and adding it again was refused as colliding
+with itself, which is the only reason anyone found out. One removed would have
+kept serving its files.
+
+**The obvious repair is wrong in the other direction**, and was committed once
+before this file said so: making the op edit the live set in place puts the new
+root on *both* sides of retarget's comparison. The table would show it and it
+would stay permanently empty. So the ops leave that object alone, the MNP path
+reloads like the loopback one always did, and the tests below check the files —
+`describe()` agreeing proves nothing about whether anything was scanned.
+"""
+
+from dataclasses import asdict
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node import ops
+from meshbay_node.config import GroupConfig, NodeConfig, RootSpec
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+from meshbay_node.roster import Roster
+
+pytestmark = pytest.mark.asyncio
+
+GROUP = "g" * 32
+
+
+async def _state(tmp_path: Path) -> tuple[dict, Roster]:
+    """A node hosting one group with two roots, as node.toml and as live state."""
+    for name in ("one", "two"):
+        (tmp_path / name).mkdir()
+
+    cfg = GroupConfig(id=GROUP, name="plop", roots=[
+        RootSpec(path=str(tmp_path / "one"), name="one"),
+        RootSpec(path=str(tmp_path / "two"), name="two"),
+    ])
+    node_cfg = NodeConfig.__new__(NodeConfig)
+    node_cfg.groups = [cfg]
+
+    conf = tmp_path / "node.toml"
+    conf.write_text(
+        f'[[groups]]\nid = "{GROUP}"\nname = "plop"\n\n'
+        f'  [[groups.roots]]\n  path = "{(tmp_path / "one").as_posix()}"\n'
+        f'  name = "one"\n\n'
+        f'  [[groups.roots]]\n  path = "{(tmp_path / "two").as_posix()}"\n'
+        f'  name = "two"\n')
+
+    live = RootSet.build([asdict(r) for r in cfg.roots])
+    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    state = {
+        "config": node_cfg,
+        "config_path": str(conf),
+        "groups_ctx": {GROUP: {"index": index, "roots": live}},
+        "roster": roster,
+        "node_user_id": "operator",
+    }
+    return state, roster
+
+
+def _live(state) -> RootSet:
+    return state["groups_ctx"][GROUP]["roots"]
+
+
+async def _indexer(state) -> DirectoryIndexer:
+    idx = DirectoryIndexer(roots=_live(state), group_id=GROUP,
+                           sk_node=Ed25519PrivateKey.generate(), gek=None)
+    await idx.initial_scan()
+    return idx
+
+
+def _rebuilt(state) -> RootSet:
+    """What a reload produces: a fresh set from the config the op just wrote."""
+    return RootSet.build([asdict(r) for r in state["config"].groups[0].roots])
+
+
+# ── What the op writes ───────────────────────────────────────────────────────
+
+async def test_adding_a_root_reaches_node_toml_and_the_ack(tmp_path):
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    try:
+        result = await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+
+        assert [r["name"] for r in result["roots"]] == ["one", "two", "uploads"]
+        assert [r.name for r in state["config"].groups[0].roots] == [
+            "one", "two", "uploads"]
+        assert "uploads" in Path(state["config_path"]).read_text()
+    finally:
+        await roster.close()
+
+
+async def test_removing_a_root_reaches_node_toml_and_the_ack(tmp_path):
+    state, roster = await _state(tmp_path)
+    try:
+        result = await ops.remove_root(state, GROUP, "two")
+        assert [r["name"] for r in result["roots"]] == ["one"]
+        assert Path(state["config_path"]).read_text().count(
+            "[[groups.roots]]") == 1
+    finally:
+        await roster.close()
+
+
+async def test_the_op_does_not_edit_the_live_set_in_place(tmp_path):
+    """
+    The property that made the original bug, and then made the first repair for
+    it wrong in the other direction.
+
+    `retarget` diffs the names it holds against the ones it is handed. Editing
+    that same object and passing it back puts a new root on both sides of the
+    comparison: nothing is scanned, and the directory shows in the table
+    permanently empty. `_reload_config_inner` diffs the same way and would
+    likewise conclude nothing had changed.
+    """
+    state, roster = await _state(tmp_path)
+    before = [r.name for r in _live(state)]
+    (tmp_path / "uploads").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        assert [r.name for r in _live(state)] == before, (
+            "add_root edited the live RootSet, which is the object retarget "
+            "diffs against — the new root would never be scanned")
+
+        await ops.remove_root(state, GROUP, "two")
+        assert [r.name for r in _live(state)] == before, (
+            "remove_root edited the live RootSet, so retarget cannot tell the "
+            "removed root's entries should go")
+    finally:
+        await roster.close()
+
+
+# ── What the node then serves ────────────────────────────────────────────────
+
+async def test_a_retarget_from_the_config_scans_the_new_root(tmp_path):
+    """
+    The half no assertion about `describe()` can reach: the files.
+
+    A root that appears in the table and holds nothing is the same bug one step
+    later, and it is what editing the live set in place would produce.
+    """
+    state, roster = await _state(tmp_path)
+    (tmp_path / "one" / "kept.txt").write_bytes(b"kept")
+    fresh = tmp_path / "uploads"
+    fresh.mkdir()
+    (fresh / "new.txt").write_bytes(b"new")
+
+    idx = await _indexer(state)
+    assert {e.name for e in idx.index.entries} == {"kept.txt"}
+    try:
+        await ops.add_root(state, GROUP, str(fresh))
+        await idx.retarget(_rebuilt(state))
+
+        assert {e.name for e in idx.index.entries} == {"kept.txt", "new.txt"}, (
+            "the added directory was not scanned — it would show in the table "
+            "and stay empty")
+        assert [r["name"] for r in idx.index.roots] == ["one", "two", "uploads"]
+    finally:
+        await roster.close()
+
+
+async def test_handing_retarget_the_edited_set_scans_nothing(tmp_path):
+    """
+    The failure mode above, demonstrated rather than described — so the reason
+    the ops leave the live set alone is checkable instead of asserted in a
+    comment. If this ever starts failing, `retarget` has changed and the rule
+    in `add_root` can be revisited.
+    """
+    state, roster = await _state(tmp_path)
+    fresh = tmp_path / "uploads"
+    fresh.mkdir()
+    (fresh / "new.txt").write_bytes(b"new")
+
+    idx = await _indexer(state)
+    try:
+        await ops.add_root(state, GROUP, str(fresh))
+        # What editing in place would have left behind.
+        _live(state).roots.append(_rebuilt(state).roots[-1])
+        await idx.retarget(_live(state))
+
+        assert {e.name for e in idx.index.entries} == set(), (
+            "retarget now scans a root it was handed on both sides of its own "
+            "diff — the constraint this file is built on has changed")
+    finally:
+        await roster.close()
+
+
+async def test_a_retarget_from_the_config_drops_a_removed_root(tmp_path):
+    """The mirror: a removed directory's files must stop being served."""
+    state, roster = await _state(tmp_path)
+    (tmp_path / "one" / "kept.txt").write_bytes(b"kept")
+    (tmp_path / "two" / "going.txt").write_bytes(b"going")
+
+    idx = await _indexer(state)
+    assert {e.name for e in idx.index.entries} == {"kept.txt", "going.txt"}
+    try:
+        await ops.remove_root(state, GROUP, "two")
+        await idx.retarget(_rebuilt(state))
+        assert {e.name for e in idx.index.entries} == {"kept.txt"}, (
+            "the removed directory's files are still being served")
+    finally:
+        await roster.close()
+
+
+# ── The invariants around them ───────────────────────────────────────────────
+
+async def test_adding_the_same_directory_twice_is_still_refused(tmp_path):
+    """A group with one path under two names indexes every file in it twice."""
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        with pytest.raises(ops.OpError):
+            await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        assert len(state["config"].groups[0].roots) == 3, (
+            "the refused add left something behind")
+        assert Path(state["config_path"]).read_text().count(
+            "[[groups.roots]]") == 3
+    finally:
+        await roster.close()
+
+
+async def test_a_second_different_root_still_lands(tmp_path):
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    (tmp_path / "incoming").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"))
+        result = await ops.add_root(state, GROUP, str(tmp_path / "incoming"),
+                                    writable=True)
+        assert [r["name"] for r in result["roots"]] == [
+            "one", "two", "uploads", "incoming"]
+        assert result["roots"][-1]["writable"] is True
+    finally:
+        await roster.close()
+
+
+async def test_updating_flags_may_edit_the_live_set(tmp_path):
+    """
+    The exception, and why it is one: `writable` and `removable` change nothing
+    about which files exist, so there is nothing for retarget to scan or drop.
+    Editing in place is what makes the flag true for the upload handler on the
+    very next request, which is synchronous and reads the live set.
+    """
+    state, roster = await _state(tmp_path)
+    state["config"] = SimpleNamespace(groups=state["config"].groups)
+    try:
+        result = await ops.update_root(state, GROUP, "two",
+                                       writable=True, removable=True)
+        live = _live(state).by_name("two")
+        assert live.writable is True and live.removable is True
+        assert result["roots"] == _live(state).describe()
+    finally:
+        await roster.close()
+
+
+async def test_the_file_on_disk_and_the_config_in_memory_agree(tmp_path):
+    """
+    A reload re-reads the file, so a config edited in memory but not on disk is
+    undone by the next restart — and one written to disk but not in memory
+    makes the *next* op validate against a stale picture.
+    """
+    state, roster = await _state(tmp_path)
+    (tmp_path / "uploads").mkdir()
+    try:
+        await ops.add_root(state, GROUP, str(tmp_path / "uploads"),
+                           writable=True)
+        await ops.remove_root(state, GROUP, "one")
+
+        import tomllib
+        on_disk = tomllib.loads(Path(state["config_path"]).read_text())
+        disk_paths = [str(r["path"]) for r in on_disk["groups"][0]["roots"]]
+        memory_paths = [Path(r.path).as_posix()
+                        for r in state["config"].groups[0].roots]
+        assert disk_paths == memory_paths
+        assert Path(state["config_path"]).read_text().count(
+            "[[groups.roots]]") == 2
+    finally:
+        await roster.close()
+
+
+# ── The seam that was actually broken ────────────────────────────────────────
+
+async def test_the_mnp_path_reloads_like_the_loopback_one(tmp_path):
+    """
+    The two front doors, doing the same thing.
+
+    `ui/app.py` has always fired the daemon's `reload_fn` after a root op.
+    `_retarget_indexer` did not — it re-pointed the indexer at the live set
+    instead, which is the object the ops leave alone, so nothing happened at
+    all. That divergence *is* the bug: the loopback path worked, the MNP path
+    did not, and it survived until an operator added a directory from a
+    browser.
+
+    Not awaited: a reload rescans, and a new library is minutes. The ack
+    already carries the set the node is moving to.
+    """
+    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+    state, roster = await _state(tmp_path)
+    reloaded: list[bool] = []
+
+    async def fake_reload():
+        reloaded.append(True)
+
+    state["reload_fn"] = fake_reload
+    spawned = []
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"daemon_state": state}
+    session._spawn = lambda coro: spawned.append(coro)
+    try:
+        await session._retarget_indexer(GROUP)
+        assert spawned, "the MNP path did not ask the daemon to reload"
+        await spawned[0]
+        assert reloaded == [True]
+    finally:
+        await roster.close()
+
+
+async def test_without_a_daemon_it_still_retargets(tmp_path):
+    """
+    A context assembled by hand — a harness, or a test — has no `reload_fn`.
+    Falling through to a direct retarget keeps those working, and is correct
+    precisely because the ops no longer edit the set being passed.
+    """
+    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+    state, roster = await _state(tmp_path)
+    fresh = tmp_path / "uploads"
+    fresh.mkdir()
+    (fresh / "new.txt").write_bytes(b"new")
+    idx = await _indexer(state)
+    state["indexers"] = {GROUP: idx}
+
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"daemon_state": state}
+    try:
+        await ops.add_root(state, GROUP, str(fresh))
+        # What a reload would have installed, done by hand here.
+        state["groups_ctx"][GROUP]["roots"] = _rebuilt(state)
+        await session._retarget_indexer(GROUP)
+        assert {e.name for e in idx.index.entries} == {"new.txt"}
+    finally:
+        await roster.close()
diff --git a/packages/meshbay-node/tests/test_root_paths_are_operator_only.py b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py
new file mode 100644
index 0000000..080d4be
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_paths_are_operator_only.py
@@ -0,0 +1,114 @@
+"""
+Where a directory lives on the operator's disk is theirs, not the group's.
+
+`RootSet.describe()` feeds two very different audiences. The index payload goes
+to every member, and has always deliberately carried no paths — a member is
+told what exists and whether it is readable, not that the library sits in
+`/media/<the operator's name>/BACKUP2`. The loopback API answers the operator
+themselves, over a channel that already requires being on their machine with
+the run token, where the path is exactly what they are asking for.
+
+`meshbay-node root list` printed `?` for every directory because it read a
+field the member form omits. Nothing caught it: the CLI reads a dict, the
+payload is a dict, and neither end says what keys it owes the other.
+
+Both halves matter and they pull opposite ways, so both are asserted here — a
+test that only checked the operator gets paths would be satisfied by putting
+them in the member payload too.
+"""
+
+import inspect
+import re
+from pathlib import Path
+
+
+from meshbay_node import daemon as daemon_mod
+from meshbay_node import ops
+from meshbay_node.roots import RootSet
+
+
+def _roots(tmp_path: Path) -> RootSet:
+    for name in ("Films", "Albums"):
+        (tmp_path / name).mkdir()
+    return RootSet.build([
+        {"path": str(tmp_path / "Films"), "writable": True},
+        {"path": str(tmp_path / "Albums"), "removable": True},
+    ])
+
+
+# ── The member's half ────────────────────────────────────────────────────────
+
+def test_the_default_form_carries_no_path(tmp_path):
+    described = _roots(tmp_path).describe()
+    assert described, "no roots described"
+    assert not any("path" in d for d in described), (
+        "the index payload every member receives would carry the operator's "
+        "filesystem layout")
+
+
+def test_the_default_form_still_says_what_a_member_needs(tmp_path):
+    """The counter-property: dropping the path must not drop the rest."""
+    described = _roots(tmp_path).describe()
+    for d in described:
+        assert set(d) >= {"name", "kind", "available", "writable",
+                          "removable", "ejected"}
+
+
+def test_the_index_payload_is_built_without_paths():
+    """
+    Read from the source, because the alternative is asserting it about a
+    payload built by a test rather than by the node.
+    """
+    from meshbay_node.indexer import indexer as indexer_mod
+    source = inspect.getsource(indexer_mod)
+    for call in re.findall(r"roots\.describe\([^)]*\)", source):
+        assert "with_paths" not in call, (
+            f"the indexer builds the member-facing roots table as {call} — "
+            f"that payload goes to everyone in the group")
+
+
+# ── The operator's half ──────────────────────────────────────────────────────
+
+def test_the_operator_form_carries_the_path(tmp_path):
+    described = _roots(tmp_path).describe(with_paths=True)
+    assert all(d.get("path") for d in described)
+    assert described[0]["path"] == str(tmp_path / "Films")
+
+
+def test_the_loopback_api_asks_for_paths():
+    """
+    `list_groups` answers the operator's own channel, and the CLI's `root list`
+    prints what it returns. Asking for the member form there is what printed a
+    column of question marks.
+    """
+    source = inspect.getsource(ops.list_groups)
+    assert "describe(with_paths=True)" in source, (
+        "list_groups uses the member form, so every path it reports is missing")
+
+
+def test_the_cli_only_reads_fields_the_payload_carries():
+    """
+    The gap this whole file exists for. The CLI reads a dict and the API
+    returns a dict; nothing between them says which keys are owed, so a name
+    that is simply absent prints as a placeholder and looks like a node
+    problem.
+    """
+    source = inspect.getsource(daemon_mod.main)
+    start = source.index('if args.command == "root":')
+    block = source[start:source.index('if args.command == "operator":', start)]
+
+    read = set(re.findall(r"r\.get\(['\"](\w+)['\"]", block))
+    read |= set(re.findall(r"r\[['\"](\w+)['\"]\]", block))
+    assert read, "the root CLI no longer reads the payload this way"
+
+    class _Any:
+        path = Path("/tmp/x")
+        name = "x"
+        kind = "generic"
+        writable = removable = ejected = False
+        available = True
+
+    offered = set(RootSet(roots=[_Any()]).describe(with_paths=True)[0])
+    assert read <= offered, (
+        f"the `root` CLI reads keys the loopback payload does not carry: "
+        f"{sorted(read - offered)}")
diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py
new file mode 100644
index 0000000..7eb75fd
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_writable_policy.py
@@ -0,0 +1,246 @@
+"""
+Who may write to the operator's disk, now that RO/RW on the root decides it.
+
+This replaces `test_member_upload_policy.py`. The old model had two orthogonal
+controls — one root designated as the upload target, and a group-wide
+`member_upload` switch — and collapsed into one property per root: `writable`.
+The properties worth keeping from the old file survive the change unaltered:
+
+* the interface hiding a control is a courtesy to the people who are not
+  trying; **the node refusing is the part that holds** against someone who is.
+  A member with an old tab open, or one speaking MNP directly, gets the same
+  answer. That half is pinned in `test_security_regressions.py`, next to the
+  overwrite properties it belongs with;
+* the setting is changed by a **signed** operator instruction, or it is a
+  suggestion any member can undo;
+* it is stored on the **node**, never the hub. A hub that could decide who
+  writes to the operator's disk would have authority over the node.
+
+And one that is new: the *old* message must no longer be able to change
+anything. A deprecated instruction that still works is not deprecated, and this
+one would reopen uploads group-wide.
+"""
+
+import base64
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import RootSet
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *,
+             writable: bool = True,
+             operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": RootSet.build([{"path": str(shared_root), "writable": writable}]),
+        "index": index,
+        "sk_node": index.sk_node,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = "g" * 32
+    session._user_id = user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _upload(session, filename="clip.mp4", body=b"bytes"):
+    session._do_file_upload({
+        "filename": filename, "dir": "shared",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(body).decode(),
+    })
+
+
+def _uploads_dir(session) -> Path:
+    # The root itself: the `uploads/` subdirectory the node used to create is
+    # gone (see test_security_regressions._uploads_dir for why).
+    return session._ctx["roots"].roots[0].path
+
+
+# ── The door, not the button ─────────────────────────────────────────────────
+
+async def test_a_member_cannot_upload_to_a_read_only_root(tmp_path):
+    session = _session(tmp_path, "member-1", writable=False)
+    _upload(session)
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not (_uploads_dir(session) / "clip.mp4").exists()
+
+
+async def test_members_upload_normally_to_a_writable_root(tmp_path):
+    session = _session(tmp_path, "member-1", writable=True)
+    _upload(session)
+
+    assert not [m for m in session.sent if m.get("type") == "error"]
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_read_only_binds_the_operator_too(tmp_path):
+    """
+    The old model exempted the operator, because the switch was about *members*.
+    RO is about the directory: a published library is read-only for everyone, and
+    an exception for admin authority is how a rule turns into a default.
+    """
+    session = _session(tmp_path, "the-operator", writable=False,
+                       operator="the-operator")
+    session._is_node_admin = lambda: True
+    _upload(session)
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+
+
+async def test_a_member_cannot_create_a_folder_in_a_read_only_root(tmp_path):
+    """
+    Read-only has to mean read-only for every way of writing, not just for
+    files. `_do_file_upload` gained this check with the RO/RW model and
+    `_do_dir_create` did not, so a member refused a file in a published library
+    could still leave empty directories all through it.
+
+    Creating a folder stays unprivileged — the node's own words: "a member who
+    can add a file can organise where it goes". What changed is that it now
+    requires the same root to be writable that adding the file would have.
+    """
+    session = _session(tmp_path, "member-1", writable=False)
+    session._do_dir_create({"dir": "shared", "name": "New folder"})
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not (tmp_path / "shared" / "New folder").exists()
+
+
+async def test_a_member_can_create_a_folder_in_a_writable_root(tmp_path):
+    """The counter-property: it must stay unprivileged where it is allowed."""
+    session = _session(tmp_path, "member-1", writable=True)
+    session._do_dir_create({"dir": "shared", "name": "New folder"})
+
+    assert not [m for m in session.sent if m.get("type") == "error"]
+    assert (tmp_path / "shared" / "New folder").is_dir()
+
+
+async def test_an_ejected_root_refuses_a_new_folder(tmp_path):
+    """Writing to a drive somebody has their hand on, one level up from a file."""
+    session = _session(tmp_path, "member-1", writable=True)
+    roots = session._ctx["roots"]
+    roots.roots[0].ejected = True
+    roots.roots[0].available = False
+
+    session._do_dir_create({"dir": "shared", "name": "New folder"})
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_unavailable"
+    assert not (tmp_path / "shared" / "New folder").exists()
+
+
+# ── Signed, or it is a suggestion ────────────────────────────────────────────
+
+def _capture_challenges(session) -> list[tuple[str, str]]:
+    issued: list[tuple[str, str]] = []
+
+    def issue(op, subject, **kw):
+        issued.append((op, subject))
+
+    session._issue_admin_challenge = issue
+    session._has_admin_authority = lambda: True
+    return issued
+
+
+async def test_changing_a_roots_flags_needs_a_signature(tmp_path):
+    """The flags are not applied by the request — only by the signed response."""
+    session = _session(tmp_path, "the-operator", operator="the-operator")
+    issued = _capture_challenges(session)
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": False})
+
+    assert [op for op, _ in issued] == [OP_ROOT_UPDATE]
+    assert session._ctx["roots"].roots[0].writable is True, (
+        "applied before it was signed")
+
+
+async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
+    """
+    The operator is shown the subject before signing, so it has to say what will
+    be true afterwards. "shared" alone would have them authorize a change they
+    cannot see the direction of.
+    """
+    session = _session(tmp_path, "op", operator="op")
+    issued = _capture_challenges(session)
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": True, "removable": True})
+
+    assert issued == [(OP_ROOT_UPDATE, "shared:rw=on,rem=on")]
+
+
+async def test_eject_and_plug_are_signed_too(tmp_path):
+    """
+    Hiding a group's whole library from every member is not a lesser act than
+    changing a flag. An unsigned one would let any member black out a group.
+    """
+    session = _session(tmp_path, "op", operator="op")
+    issued = _capture_challenges(session)
+
+    session._do_root_eject({"group_id": "g" * 32, "root_name": "shared"})
+    session._do_root_plug({"group_id": "g" * 32, "root_name": "shared"})
+
+    assert issued == [(OP_ROOT_EJECT, "shared"), (OP_ROOT_PLUG, "shared")]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    """
+    An unpaired node has no key to check a signature against, so the challenge
+    is never issued rather than issued and then unverifiable.
+    """
+    session = _session(tmp_path, "member-1")
+    issued = _capture_challenges(session)
+    session._has_admin_authority = lambda: False
+
+    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
+                             "writable": True})
+
+    assert issued == []
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── The deprecated message must not still work ───────────────────────────────
+
+async def test_the_old_member_upload_message_changes_nothing(tmp_path):
+    """
+    MNP still parses `member_upload` so an old client gets an answer instead of
+    a dropped request. What it must not do is act: this instruction could
+    reopen uploads for a whole group, and a client old enough to send it is
+    exactly one that knows nothing about read-only roots.
+    """
+    session = _session(tmp_path, "member-1", writable=False)
+    session._has_admin_authority = lambda: True
+    issued = _capture_challenges(session)
+
+    session._do_member_upload({"allowed": True})
+
+    assert issued == [], "a deprecated instruction asked to be signed"
+    assert session._ctx["roots"].roots[0].writable is False
+    acks = [m for m in session.sent if m.get("type") == MNP.MEMBER_UPLOAD_ACK]
+    assert acks and acks[0].get("deprecated") is True
+
+    # And the door is still shut.
+    _upload(session)
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
diff --git a/packages/meshbay-node/tests/test_roots.py b/packages/meshbay-node/tests/test_roots.py
index 1beb220..505091b 100644
--- a/packages/meshbay-node/tests/test_roots.py
+++ b/packages/meshbay-node/tests/test_roots.py
@@ -108,31 +108,61 @@ def test_a_sibling_with_a_shared_prefix_is_fine(tmp_path):
     assert roots.names == ["Media", "Media2"]
 
 
-# ── Uploads ──────────────────────────────────────────────────────────────────
+# ── Writable roots ───────────────────────────────────────────────────────────
 
-def test_a_single_root_receives_uploads_without_being_asked(tmp_path):
+def test_a_root_is_read_only_unless_it_says_otherwise(tmp_path):
+    """
+    The default is the safe one. An operator who shares a directory has not
+    thereby agreed to let anyone write into it, and the version of this that
+    guessed — one root, so it must be the upload target — meant adding a
+    second directory silently changed what the first one was.
+    """
     (tmp_path / "Media").mkdir()
     roots = RootSet.build([_spec(tmp_path / "Media")])
-    assert roots.upload_root is roots.roots[0]
+    assert roots.roots[0].writable is False
+    assert roots.writable_roots == []
+
+
+def test_several_roots_can_be_writable_at_once(tmp_path):
+    (tmp_path / "A").mkdir()
+    (tmp_path / "B").mkdir()
+    (tmp_path / "C").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "A", writable=True),
+                           _spec(tmp_path / "B"),
+                           _spec(tmp_path / "C", writable=True)])
+    assert [r.name for r in roots.writable_roots] == ["A", "C"]
 
 
-def test_several_roots_and_no_designation_means_no_uploads(tmp_path):
+def test_a_fully_read_only_group_is_valid(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.
+    A group that only publishes is the point of the read-only model, not a
+    misconfiguration — build must not refuse it, and nothing downstream may
+    promote a root to writable to have somewhere to put an upload.
     """
     (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
+    assert roots.writable_roots == []
+    assert len(roots) == 2
 
 
-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)])
+def test_the_old_upload_flag_still_reads_as_writable(tmp_path):
+    """A node.toml written before this refactor must not change meaning."""
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", upload=True)])
+    assert roots.roots[0].writable is True
+    assert roots.describe()[0]["writable"] is True
+
+
+def test_writable_wins_over_a_leftover_upload_flag(tmp_path):
+    """
+    A config carrying both is one a migration touched. `writable` is the field
+    the operator's tooling writes now, so it is the one that decides — reading
+    the legacy field there would undo the migration on the next load.
+    """
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", upload=True, writable=False)])
+    assert roots.roots[0].writable is False
 
 
 # ── Resolution ───────────────────────────────────────────────────────────────
@@ -236,18 +266,35 @@ def test_availability_follows_the_directory(tmp_path):
 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")])
+    roots = RootSet.build([_spec(tmp_path / "Media", writable=True),
+                           _spec(tmp_path / "Music", kind="audio",
+                                 removable=True)])
     described = roots.describe()
     assert described == [
-        {"name": "Media", "kind": "generic", "available": True, "upload": True},
-        {"name": "Music", "kind": "audio", "available": True, "upload": False},
+        {"name": "Media", "kind": "generic", "available": True,
+         "writable": True, "removable": False, "ejected": False,
+         "upload": True},
+        {"name": "Music", "kind": "audio", "available": True,
+         "writable": False, "removable": True, "ejected": False,
+         "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)
 
 
+def test_describe_still_carries_upload_for_mnp_1_0_clients(tmp_path):
+    """
+    `upload` is `writable` under its old name, kept because an MNP 1.0 client
+    reads no other field and would otherwise decide the group takes no uploads
+    at all. It is derived, never stored — the two can never disagree.
+    """
+    (tmp_path / "Media").mkdir()
+    roots = RootSet.build([_spec(tmp_path / "Media", writable=True)])
+    described = roots.describe()[0]
+    assert described["upload"] == described["writable"] is True
+
+
 # ── SAFE_UPLOAD_NAME ────────────────────────────────────────────────────────
 
 def test_safe_name_accepts_unicode_letters():
diff --git a/packages/meshbay-node/tests/test_scan_settings_policy.py b/packages/meshbay-node/tests/test_scan_settings_policy.py
index 719b988..94f4421 100644
--- a/packages/meshbay-node/tests/test_scan_settings_policy.py
+++ b/packages/meshbay-node/tests/test_scan_settings_policy.py
@@ -2,7 +2,7 @@
 The operator can tune how often the indexer's reconciliation backstop runs,
 and how long it waits after a file's last write before hashing it.
 
-Same shape as test_apps_enabled_policy.py / test_member_upload_policy.py:
+Same shape as test_apps_enabled_policy.py / test_root_writable_policy.py:
 changed by a signed operator instruction, stored on the node rather than the
 hub. Unlike those two, there is also a *live* DirectoryIndexer object to
 update — see test_set_scan_settings_updates_the_live_indexer below.
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 7f71da5..1a318f7 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 meshbay_node.roots import RootSet
 from conftest import one_root
 from meshbay_node.transport.webrtc_server import WebRTCPeerSession
 
@@ -132,14 +133,22 @@ def test_the_node_never_generates_a_name_it_would_refuse(tmp_path):
 
 def _uploads_dir(session) -> Path:
     """
-    Where this session's uploads land: uploads/ inside the group's upload root.
+    Where an unaddressed upload lands: the first writable root itself.
+
+    There is no `uploads/` subdirectory any more. It was the last of v5's
+    quarantine — the per-user layer went on 2026-08-14 — and it went for the
+    same reason: a folder appearing beside the operator's library because
+    somebody sent a file is the node deciding how their disk is arranged. The
+    protections that made the quarantine worth having are the allowlist, the
+    size cap, the chunk ordering and the no-overwrite rule, and every one of
+    them is asserted below, unchanged.
 
     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"
+    writable = session._ctx["roots"].writable_roots
+    assert writable, "the fixture must give the group a writable root"
+    return writable[0].path
 
 
 def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
@@ -175,7 +184,6 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path):
     """
     victim = _session(tmp_path, "victim-user")
     uploads = _uploads_dir(victim)
-    uploads.mkdir()
     original = uploads / "important.mp4"
     original.write_bytes(b"operator's original content")
 
@@ -225,22 +233,157 @@ def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
     assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}"
 
 
-def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
+def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
+    """
+    The destination is now the folder the sender is looking at, which means the
+    client does choose it — and the whole of what keeps that safe is that the
+    choice is *resolved against the group's own roots* rather than joined to
+    one.
+
+    `RootSet.resolve()` refuses `..`, absolute segments and anything whose
+    resolved form escapes its root, symlinks included. So "which of this
+    group's folders" is answerable by a member and "which path on the
+    operator's disk" is not.
+    """
+    session = _session(tmp_path, "user-1")
+    (session._ctx["roots"].roots[0].path / "sub").mkdir()
+    before = set(tmp_path.rglob("*"))
+
+    for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc",
+                "nope", "shared/missing"):
+        session.sent.clear()
+        session._do_file_upload({
+            "filename": "note.txt", "dir": bad,
+            "chunk_index": 0, "total_chunks": 1,
+            "data": base64.b64encode(b"x").decode(),
+        })
+        refusal = [m for m in session.sent if m.get("type") == "error"]
+        assert refusal, f"{bad!r} was accepted"
+        assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad
+
+    assert set(tmp_path.rglob("*")) == before, "a refused upload still wrote"
+
+
+def test_an_upload_lands_in_the_folder_it_names(tmp_path):
+    """
+    And in that folder itself — the `uploads/` subdirectory the node used to
+    create is gone. Somebody dropping a file into the folder they are looking
+    at expects it to be in that folder.
+    """
+    session = _session(tmp_path, "user-1")
+    root = session._ctx["roots"].roots[0]
+    (root.path / "Albums").mkdir()
+
+    session._do_file_upload({
+        "filename": "note.txt", "dir": f"{root.name}/Albums",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    assert (root.path / "Albums" / "note.txt").read_bytes() == b"x"
+    assert not (root.path / "Albums" / "uploads").exists(), (
+        "the node invented a subdirectory in the operator's library")
+    assert not (root.path / "uploads").exists()
+
+
+def test_an_upload_goes_to_the_root_it_names(tmp_path):
+    """
+    With two writable roots there is no defensible default, and the client is
+    the only party that knows which directory the person is looking at. The
+    node picking one meant a file uploaded from a folder on screen landed in a
+    different one — the same "uploads went somewhere else" the single upload
+    root was never allowed to guess about.
+    """
+    media = tmp_path / "Media"
+    incoming = tmp_path / "Incoming"
+    media.mkdir()
+    incoming.mkdir()
+    session = _session(tmp_path, "user-1")
+    session._ctx["roots"] = RootSet.build([
+        {"path": str(media), "writable": True},
+        {"path": str(incoming), "writable": True},
+    ])
+
+    session._do_file_upload({
+        "filename": "note.txt", "dir": "Incoming",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    assert (incoming / "note.txt").read_bytes() == b"x"
+    assert not (media / "note.txt").exists(), "it went to the first root instead"
+
+
+def test_a_read_only_root_refuses_an_upload(tmp_path):
+    """
+    RO is the mechanism now, not a hidden button. It binds the operator too:
+    "read-only for everyone" is what makes a published library one, and an
+    exception for whoever happens to hold admin authority is the sort of
+    carve-out that later reads as the rule.
+    """
+    published = tmp_path / "Published"
+    published.mkdir()
+    session = _session(tmp_path, "user-1")
+    session._ctx["roots"] = RootSet.build([{"path": str(published)}])
+    session._is_node_admin = lambda: True
+
+    session._do_file_upload({
+        "filename": "note.txt", "dir": "Published",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_read_only"
+    assert not (published / "note.txt").exists()
+
+
+def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
+    """
+    An MNP 1.0 client names no root, so the node falls back to the first
+    writable one. There isn't one here, and the fallback must refuse rather
+    than write into whatever root happens to come first.
+    """
+    published = tmp_path / "Published"
+    published.mkdir()
+    session = _session(tmp_path, "user-1")
+    session._ctx["roots"] = RootSet.build([{"path": str(published)}])
+
+    session._do_file_upload({
+        "filename": "note.txt",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "no_writable_root"
+    assert not (published / "note.txt").exists()
+
+
+def test_an_ejected_root_refuses_an_upload(tmp_path):
     """
-    Uploads land in uploads/, chosen by the node. A client that names somewhere
-    else — or nowhere at all — changes nothing, so the traversal surface that a
-    client-chosen destination would open does not exist on this path.
+    Writing to a drive somebody has their hand on is the thing eject exists to
+    stop. `writable` is still true — that is configuration — so availability
+    has to be checked separately, which is what an earlier version conflated.
     """
+    usb = tmp_path / "USB"
+    usb.mkdir()
     session = _session(tmp_path, "user-1")
+    roots = RootSet.build([{"path": str(usb), "writable": True,
+                            "removable": True}])
+    roots.roots[0].ejected = True
+    roots.roots[0].available = False
+    session._ctx["roots"] = roots
 
     session._do_file_upload({
-        "filename": "note.txt", "dir": "../../etc",
+        "filename": "note.txt", "dir": "USB",
         "chunk_index": 0, "total_chunks": 1,
         "data": base64.b64encode(b"x").decode(),
     })
 
-    assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x"
-    assert not (tmp_path / "etc").exists()
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "root_unavailable"
+    assert not (usb / "note.txt").exists()
 
 
 def test_two_members_can_send_the_same_filename(tmp_path):
diff --git a/packages/meshbay-node/tests/test_startup_scan_enrichment.py b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
index 65b9728..d2cbc3e 100644
--- a/packages/meshbay-node/tests/test_startup_scan_enrichment.py
+++ b/packages/meshbay-node/tests/test_startup_scan_enrichment.py
@@ -60,8 +60,9 @@ async def test_a_file_already_on_disk_at_startup_gets_enrichment_scheduled(tmp_p
         data_dir=tmp_path / "data",
     )
     class _StubRoster:
-        async def video_root(self, group_id):
-            return "shared"  # the root itself, i.e. "enrich the whole thing"
+        async def app_directories(self, group_id, app_key):
+            # The root itself, i.e. "enrich the whole thing".
+            return ["shared"] if app_key == "video" else []
 
     daemon = NodeDaemon(config)
     daemon._broadcast_coalesce_secs = 0.01  # real value would make this test wait 0.5s
diff --git a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
index 9cfb819..b88ecaf 100644
--- a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
+++ b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py
@@ -99,7 +99,7 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
 
     daemon = await _make_daemon(tmp_path, shared, group_id)
     try:
-        await daemon._roster.set_video_root(group_id, "shared/Movies", set_by="op")
+        await daemon._roster.set_app_directories(group_id, "video", ["shared/Movies"], set_by="op")
         indexer = DirectoryIndexer(
             roots=one_root(shared), group_id=group_id,
             sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
@@ -139,8 +139,10 @@ async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path):
         # this file all along.
         state = {
             "roster": daemon._roster,
-            "groups_ctx": {group_id: {}},
-            "enrich_video_root_fn": daemon._enrich_video_root_now,
+            # Real roots, because ops now refuses a directory that is not
+            # inside one — the per-app setters this replaced validated nothing.
+            "groups_ctx": {group_id: {"roots": one_root(shared)}},
+            "enrich_app_dirs_fns": {"video": daemon._enrich_video_root_now},
         }
         await ops.set_video_root(state, group_id, "shared/Movies")
         await asyncio.sleep(0.05)  # let the fire-and-forget sweep actually run
diff --git a/packages/meshbay-node/tests/test_video_root_policy.py b/packages/meshbay-node/tests/test_video_root_policy.py
index 8cc1540..8d8c45a 100644
--- a/packages/meshbay-node/tests/test_video_root_policy.py
+++ b/packages/meshbay-node/tests/test_video_root_policy.py
@@ -126,16 +126,19 @@ async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
     roster = Roster(db_path=tmp_path / "roster.db")
     await roster.open()
     try:
-        assert await roster.video_root("g1") == "", "absent must mean the whole group index"
-        await roster.set_video_root("g1", "shared/Movies", set_by="op")
-        assert await roster.video_root("g1") == "shared/Movies"
+        assert await roster.app_directories("g1", "video") == [], (
+            "absent must mean nothing configured")
+        await roster.set_app_directories("g1", "video", ["shared/Movies"],
+                                         set_by="op")
+        assert await roster.app_directories("g1", "video") == ["shared/Movies"]
     finally:
         await roster.close()
 
     reopened = Roster(db_path=tmp_path / "roster.db")
     await reopened.open()
     try:
-        assert await reopened.video_root("g1") == "shared/Movies"
-        assert await reopened.video_root("g2") == "", "one group's setting must not answer for another"
+        assert await reopened.app_directories("g1", "video") == ["shared/Movies"]
+        assert await reopened.app_directories("g2", "video") == [], (
+            "one group's setting must not answer for another")
     finally:
         await reopened.close()
diff --git a/packages/meshbay-node/tests/test_windows_root_shapes.py b/packages/meshbay-node/tests/test_windows_root_shapes.py
new file mode 100644
index 0000000..5c5da25
--- /dev/null
+++ b/packages/meshbay-node/tests/test_windows_root_shapes.py
@@ -0,0 +1,149 @@
+"""
+The root model against the shapes Windows produces.
+
+CLAUDE.md is explicit that exFAT/NTFS and Windows are the common case, not an
+edge case: most operators are expected to share from an external drive on
+Windows. The RO/RW refactor added two booleans and a config rewriter, and the
+booleans are path-independent — but the rewriter, the name derivation and the
+collision check all touch paths, and none of them has ever run on Windows here.
+
+What this can check without Windows is the *shape* work: drive letters through
+`as_posix()`, a path with no basename to derive a name from, UNC, and a
+case-insensitive collision. `PureWindowsPath` is used deliberately — the plain
+`Path` on this machine is a `PosixPath`, where a backslash is an ordinary
+filename character, which is the mistake that made
+`test_a_backslash_path_written_into_node_toml_stays_parseable` fail everywhere
+but the platform it was written for.
+
+What it cannot check is the filesystem itself: `ReadDirectoryChangesW` dropping
+events under load, `MAX_PATH`, and whether an eject actually lets a drive be
+removed. Those need a person with Windows, and §4.4 of the refactor plan is
+where that is written down.
+"""
+
+import os
+import tempfile
+import tomllib
+from pathlib import Path, PureWindowsPath
+
+import pytest
+
+from meshbay_node.roots import RootError, RootSet, derive_name
+
+BS = chr(92)
+
+
+# ── Paths into node.toml ─────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("raw,expected", [
+    (f"D:{BS}Movies", "D:/Movies"),
+    (f"E:{BS}Music{BS}Albums", "E:/Music/Albums"),
+    (f"C:{BS}Users{BS}alice{BS}Media", "C:/Users/alice/Media"),
+    (f"{BS}{BS}server{BS}share{BS}Media", "//server/share/Media"),
+])
+def test_a_windows_path_survives_the_config_file(raw, expected):
+    """
+    `ops` writes `as_posix()` into a TOML basic string, where a raw backslash
+    is an escape — `\\U` and `\\a` are the ones that bite — so the file would
+    not parse at all. pathlib reads the forward-slash form back on Windows.
+    """
+    posix = PureWindowsPath(raw).as_posix()
+    assert posix == expected
+    parsed = tomllib.loads(f'path = "{posix}"\n')
+    assert parsed["path"] == expected
+
+
+def test_a_raw_windows_path_would_not_parse():
+    """The counter-property: without `as_posix()` there is no config file."""
+    with pytest.raises(tomllib.TOMLDecodeError):
+        tomllib.loads(f'path = "C:{BS}Users{BS}alice{BS}Media"\n')
+
+
+# ── Naming a drive ───────────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("raw,name", [
+    (f"D:{BS}Movies", "Movies"),
+    (f"E:{BS}Music{BS}Albums", "Albums"),
+    (f"{BS}{BS}server{BS}share{BS}Media", "Media"),
+])
+def test_a_name_is_derived_from_the_last_segment(raw, name):
+    assert PureWindowsPath(raw).name == name
+
+
+@pytest.mark.parametrize("raw", [f"D:{BS}", f"E:{BS}", f"{BS}{BS}server{BS}share"])
+def test_a_drive_root_has_no_name_to_derive(raw):
+    """
+    Sharing a whole drive is an ordinary thing to do on Windows and there is
+    nothing to call it, so the operator has to say. Refused with that as the
+    message rather than named "" or "D:".
+    """
+    p = PureWindowsPath(raw)
+    if p.name:
+        pytest.skip(f"{raw!r} has a basename on this platform")
+    with pytest.raises(RootError, match="explicit"):
+        derive_name(p)
+
+
+def test_naming_it_explicitly_works():
+    with tempfile.TemporaryDirectory() as d:
+        roots = RootSet.build([{"path": d, "name": "Films"}])
+        assert roots.names == ["Films"]
+
+
+# ── Case, which Windows makes real ───────────────────────────────────────────
+
+def test_two_roots_differing_only_in_case_are_refused():
+    """
+    On NTFS and exFAT `Movies` and `MOVIES` are the same directory to the
+    filesystem and two roots to a case-sensitive comparison — which would index
+    one tree twice, and make deleting a file from one copy break the other.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "Movies"))
+        os.makedirs(os.path.join(d, "other"))
+        with pytest.raises(RootError, match="regard to case"):
+            RootSet.build([
+                {"path": os.path.join(d, "Movies")},
+                {"path": os.path.join(d, "other"), "name": "MOVIES"},
+            ])
+
+
+def test_a_root_is_found_by_name_without_regard_to_case():
+    """
+    What a client sends is what a person typed or a path it split, and on
+    Windows those disagree about case routinely.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "Movies"))
+        roots = RootSet.build([{"path": os.path.join(d, "Movies")}])
+        for spelling in ("Movies", "movies", "MOVIES", "MoViEs"):
+            assert roots.by_name(spelling) is not None, spelling
+
+
+# ── The two flags ────────────────────────────────────────────────────────────
+
+def test_the_flags_do_not_touch_paths():
+    """
+    `writable` and `removable` are booleans and stay booleans on every
+    platform. Stated as a test because it is the reason the rest of the
+    refactor needed no Windows work: what did need it is above.
+    """
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "USB"))
+        roots = RootSet.build([{"path": os.path.join(d, "USB"),
+                                "writable": True, "removable": True}])
+        described = roots.describe()[0]
+        assert described["writable"] is True
+        assert described["removable"] is True
+        assert "path" not in described
+
+
+def test_an_ejected_removable_root_is_unavailable_wherever_it_runs():
+    with tempfile.TemporaryDirectory() as d:
+        os.makedirs(os.path.join(d, "USB"))
+        roots = RootSet.build([{"path": os.path.join(d, "USB"),
+                                "removable": True, "ejected": True}])
+        assert roots.roots[0].available is False
+        assert Path(roots.roots[0].path).is_dir(), (
+            "the directory is still there; `ejected` is the operator's answer, "
+            "not the filesystem's")