From ea56b8c79538323875c00db2e7006b255f7cd494 Mon Sep 17 00:00:00 2001
From: Christophe Besson <cbesson@gmail.com>
Date: Sun, 6 Sep 2026 17:48:36 +0200
Subject: fix(groups): finish Phase 1 — MNP root management, upload targets,
 eject state
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review of the Phase 1 commit found the RO/RW model sound but three paths
unfinished, each of which broke the flow the phase exists to deliver. Plus
29 test failures it introduced and no coverage for anything it added.

Uploads went to the wrong directory. The node read a `root` field on
file_upload that no client ever sent, so every upload landed in the first
writable root while the Files toolbar offered its button based on the root
being browsed — with two writable roots, uploading from one wrote into the
other. Files now names the root it is showing; Chat names one chosen in the
shell (an operator-configured directory arrives in Phase 2); the node refuses
an unknown name rather than falling back, and refuses read-only and ejected
roots by code.

Shared directories were unreachable on the web. The table read its roots
only from the loopback API, which resolves to "not available" in a browser,
so the section rendered for nobody there — while the Uploads controls it
replaced had worked — and the transport.updateRoot/ejectRoot/plugRoot methods
beside it were dead. MNP is now the path, loopback the fallback for a local
node with no live connection, and adding a root over MNP takes a typed path
since no web page can browse a remote disk.

Ejecting updated nobody's screen. transport.js resolves an admin ack against
the pending request and returns, which is right for every op whose caller
knows the value it chose; the root acks carry state only the node can compute,
so the operator who clicked Eject was the one client that never saw it happen.
And the ejected flag reached roster.db but was never read back, so a restart
undid it and the next scan read an empty mount point as an erased library.

Also: the member-upload endpoint answered 200 and did nothing (removed); the
wizard ignored the first root's RW switch; reload compared roots on name and
path, so editing writable in node.toml did nothing; the table had no path
column, which is the only thing separating two libraries sharing a basename;
apps_enabled normalisation differed between the two sides of a signed subject.

Tests: eject/plug, per-root upload refusal and the node.toml rewrite had no
coverage at all. test_member_upload_policy.py is replaced by
test_root_writable_policy.py — it tested a removed feature — and every
property worth keeping from it moved rather than being dropped.

Docs: draft-v6 structural decision 9 is annotated as superseded (the operator
can no longer have a directory only they may write to — a real capability
removed, flagged rather than hidden), the man page documents the root verb and
the RO/RW fields, and refactor-groups.md §7b records what the plan got wrong.

Suite: 41 failures before, 13 after — all 13 pre-existing on main.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011pvMdvLBG92jyhvD5pD6us
---
 packages/meshbay-node/tests/conftest.py            |  10 +-
 .../meshbay-node/tests/test_apps_enabled_policy.py |   4 +-
 packages/meshbay-node/tests/test_cli_dispatch.py   |   9 +
 .../tests/test_member_upload_policy.py             | 176 --------------
 packages/meshbay-node/tests/test_node_status.py    | 117 ++++++++-
 packages/meshbay-node/tests/test_ops.py            | 100 +++++++-
 .../meshbay-node/tests/test_root_availability.py   |  11 +-
 packages/meshbay-node/tests/test_root_eject.py     | 268 +++++++++++++++++++++
 .../tests/test_root_writable_policy.py             | 203 ++++++++++++++++
 packages/meshbay-node/tests/test_roots.py          |  81 +++++--
 .../tests/test_scan_settings_policy.py             |   2 +-
 .../tests/test_security_regressions.py             | 145 ++++++++++-
 12 files changed, 895 insertions(+), 231 deletions(-)
 delete mode 100644 packages/meshbay-node/tests/test_member_upload_policy.py
 create mode 100644 packages/meshbay-node/tests/test_root_eject.py
 create mode 100644 packages/meshbay-node/tests/test_root_writable_policy.py

(limited to 'packages/meshbay-node/tests')

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_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_cli_dispatch.py b/packages/meshbay-node/tests/test_cli_dispatch.py
index f58c020..2ba251f 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"],
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..b3f0378 100644
--- a/packages/meshbay-node/tests/test_ops.py
+++ b/packages/meshbay-node/tests/test_ops.py
@@ -12,11 +12,13 @@ call them.
 import asyncio
 import inspect
 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.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.plug_root(state, "g" * 32, "USB")
+        assert out["status"] == "plugged"
+        assert await roster.ejected_roots("g" * 32) == set()
+    finally:
+        await roster.close()
 
-    out = await ops.set_member_upload(state, "g" * 32, True)
 
-    assert out["allowed"] is True
-    assert state["groups_ctx"]["g" * 32]["member_upload"] is True
+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()
 
-    out2 = await ops.set_member_upload(state, "g" * 32, False)
 
-    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 ──────────────────────────────────────────────────────────────────
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..0ec36a4
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_eject.py
@@ -0,0 +1,268 @@
+"""
+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.
+"""
+
+import asyncio
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_node.indexer.indexer import DirectoryIndexer
+from meshbay_node.roots import RootSet
+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_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py
new file mode 100644
index 0000000..da95032
--- /dev/null
+++ b/packages/meshbay-node/tests/test_root_writable_policy.py
@@ -0,0 +1,203 @@
+"""
+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, "root": "shared",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(body).decode(),
+    })
+
+
+def _uploads_dir(session) -> Path:
+    return session._ctx["roots"].roots[0].path / "uploads"
+
+
+# ── 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).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"
+
+
+# ── 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..9db8ac1 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,14 @@ 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 this session's uploads land: uploads/ inside its first writable root.
 
     Asked of the root set rather than assembled by hand, so a test cannot pass
     while agreeing with a wrong answer the code also produced.
     """
-    root = session._ctx["roots"].upload_root
-    assert root is not None, "the fixture must designate an upload root"
-    return root.path / "uploads"
+    writable = session._ctx["roots"].writable_roots
+    assert writable, "the fixture must give the group a writable root"
+    return writable[0].path / "uploads"
 
 
 def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
@@ -227,14 +228,17 @@ def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
 
 def test_upload_ignores_any_directory_the_client_asks_for(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.
+    The destination inside a root is the node's decision, and stays so.
+
+    A client now names the *root* it is uploading into — it has to, once a group
+    can have several writable ones — but that is a name looked up in the root
+    table, never a path. Everything below the root is still chosen here, so the
+    traversal surface a client-chosen destination would open does not exist.
     """
     session = _session(tmp_path, "user-1")
 
     session._do_file_upload({
-        "filename": "note.txt", "dir": "../../etc",
+        "filename": "note.txt", "dir": "../../etc", "path": "/etc",
         "chunk_index": 0, "total_chunks": 1,
         "data": base64.b64encode(b"x").decode(),
     })
@@ -243,6 +247,131 @@ def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
     assert not (tmp_path / "etc").exists()
 
 
+@pytest.mark.parametrize("named_root", [
+    "../../etc", "/etc", "shared/../..", "Shared/uploads", "nope",
+])
+def test_a_root_name_is_looked_up_never_joined(tmp_path, named_root):
+    """
+    The name the client sends is matched against the group's root table and
+    refused when it matches nothing. A version that joined it to a path — or
+    that quietly fell back to the first writable root — would turn "which
+    directory" into either a traversal or a file on a disk the operator did
+    not intend, and the second is discovered weeks later.
+    """
+    session = _session(tmp_path, "user-1")
+    before = set(tmp_path.rglob("*"))
+
+    session._do_file_upload({
+        "filename": "note.txt", "root": named_root,
+        "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_such_root", named_root
+    assert set(tmp_path.rglob("*")) == before, f"wrote something via {named_root!r}"
+
+
+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", "root": "Incoming",
+        "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(b"x").decode(),
+    })
+
+    assert (incoming / "uploads" / "note.txt").read_bytes() == b"x"
+    assert not (media / "uploads").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", "root": "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 / "uploads").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 / "uploads").exists()
+
+
+def test_an_ejected_root_refuses_an_upload(tmp_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", "root": "USB",
+        "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_unavailable"
+    assert not (usb / "uploads").exists()
+
+
 def test_two_members_can_send_the_same_filename(tmp_path):
     """
     One shared uploads/ means collisions are ordinary — every camera produces
-- 
cgit v1.2.3