fix(groups): finish Phase 1 — MNP root management, upload targets, eject state

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

1 files changed, 268 insertions, 0 deletions

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()