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, 67 insertions, 32 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index f45101f..4fc07ad 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -36,6 +36,7 @@ from pathlib import Path
 
 import uvicorn
 
+from meshbay_common.paths import fold
 from meshbay_common import MNP_VERSION
 from meshbay_common.protocol import MNP
 from meshbay_node.audit import AuditStore
@@ -128,6 +129,11 @@ class _WsSender:
 
 # ── Daemon ────────────────────────────────────────────────────────────────────
 
+def _root_shape(roots) -> set[tuple]:
+    """What has to match for a group's roots to count as unchanged on reload."""
+    return {(r.name, str(r.path), r.writable, r.removable) for r in roots}
+
+
 class NodeDaemon:
     def __init__(self, config: Config, config_path: Path = DEFAULT_CONFIG_PATH):
         self._config = config
@@ -303,7 +309,7 @@ class NodeDaemon:
                     continue
 
                 try:
-                    roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                    roots = await self._build_roots(group_cfg)
                 except RootError as e:
                     # Configuration the operator has to fix; guessing would put
                     # a member's file on the wrong disk or index one twice.
@@ -330,7 +336,7 @@ class NodeDaemon:
                     log.info("No GEK yet for group %s — will accept first setup",
                              group_cfg.name)
 
-                # Read once at load, like member_upload/enabled_apps below —
+                # Read once at load, like enabled_apps below —
                 # kept current in place afterwards by set_scan_settings
                 # (ops.py), which updates both this indexer object directly
                 # and roster.db, so a restart picks up the same values.
@@ -347,6 +353,7 @@ class NodeDaemon:
                     sk_node=keys.sk_ed25519,
                     gek=gek,
                     on_change=self._on_index_change,
+                    on_root_ejected=self._eject_persister(group_cfg.id),
                     cache=self._index_cache,
                     reconcile_secs=scan_settings["reconcile_interval_secs"],
                     debounce_secs=scan_settings["debounce_secs"],
@@ -374,22 +381,19 @@ class NodeDaemon:
                     "note_activity": indexer.note_activity,
                     # Shown to the operator in Settings, and kept current in
                     # place by set_scan_settings (ops.py) — same reasoning as
-                    # member_upload below.
+                    # enabled_apps below.
                     "reconcile_interval_secs": scan_settings["reconcile_interval_secs"],
                     "debounce_secs": scan_settings["debounce_secs"],
                     "visibility": group_cfg.visibility,
                     # Admission policy comes from node.toml, never from the hub:
                     # a hub that could declare a group open would be handed its key.
                     "join_policy": group_cfg.join_policy,
-                    # Whether ordinary members may upload. Read once here, into
-                    # the context, because the upload handler is synchronous and
-                    # a database round trip per chunk would be absurd. The
-                    # signed operation that changes it updates this dict in
-                    # place, so the two never drift within a run.
-                    "member_upload": await self._roster.member_upload_allowed(
-                        group_cfg.id) if self._roster else True,
-                    # Same reasoning: read once at load, kept current in place
-                    # by the signed operation that changes it.
+                    # Read once at load, kept current in place by the signed
+                    # operation that changes it — the upload handler is
+                    # synchronous and a database round trip per chunk would be
+                    # absurd. (Whether a member may upload is not here any
+                    # more: it is `writable` on the root being written to,
+                    # which the RootSet above already carries.)
                     "enabled_apps": await self._roster.enabled_apps(
                         group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS),
                     # Which folder is the Videos app's entry point for this
@@ -745,14 +749,16 @@ class NodeDaemon:
             if not ctx:
                 continue
             try:
-                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                roots = await self._build_roots(group_cfg)
             except RootError as e:
                 log.error("Group %r: %s — keeping the roots already loaded",
                           group_cfg.name, e)
                 continue
-            before = {(r.name, str(r.path)) for r in ctx["roots"]}
-            after = {(r.name, str(r.path)) for r in roots}
-            if before == after:
+            # `writable` and `removable` are in the comparison because an
+            # operator editing node.toml by hand and reloading is a supported
+            # way to change them, and a set compared on name and path alone
+            # reports "nothing changed" for exactly that edit.
+            if _root_shape(ctx["roots"]) == _root_shape(roots):
                 continue
             roots.refresh_availability()
             indexer = next((i for i in self._indexers
@@ -786,7 +792,7 @@ class NodeDaemon:
                 continue
 
             try:
-                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                roots = await self._build_roots(group_cfg)
             except RootError as e:
                 log.error("New group %r: %s — skipping", group_cfg.name, e)
                 continue
@@ -812,6 +818,7 @@ class NodeDaemon:
                 sk_node=sk_ed,
                 gek=gek,
                 on_change=self._on_index_change,
+                on_root_ejected=self._eject_persister(group_cfg.id),
                 cache=self._index_cache,
                 reconcile_secs=scan_settings["reconcile_interval_secs"],
                 debounce_secs=scan_settings["debounce_secs"],
@@ -844,9 +851,6 @@ class NodeDaemon:
                 "debounce_secs": scan_settings["debounce_secs"],
                 "visibility": group_cfg.visibility,
                 "join_policy": group_cfg.join_policy,
-                "member_upload": (
-                    await self._roster.member_upload_allowed(group_cfg.id)
-                    if self._roster else True),
                 "enabled_apps": (
                     await self._roster.enabled_apps(group_cfg.id)
                     if self._roster else list(Roster.DEFAULT_APPS)),
@@ -1052,6 +1056,35 @@ class NodeDaemon:
             log.debug("Index progress pushed to %d peer(s) for group %s",
                       pushed, group_id[:8])
 
+    async def _build_roots(self, group_cfg) -> RootSet:
+        """
+        Build a group's RootSet from node.toml, with the ejected state restored.
+
+        node.toml carries configuration (`writable`, `removable`); the roster
+        carries the runtime answer to "is this drive ejected right now". They
+        are merged here, in the one place every caller goes through, because a
+        root that quietly comes back available across a restart is exactly the
+        surprise unplug that eject exists to survive.
+        """
+        specs = [asdict(r) for r in group_cfg.roots]
+        if self._roster:
+            ejected = await self._roster.ejected_roots(group_cfg.id)
+            if ejected:
+                for spec in specs:
+                    name = spec.get("name") or Path(spec.get("path", "")).name
+                    if fold(name) in ejected:
+                        spec["ejected"] = True
+        return RootSet.build(specs)
+
+    def _eject_persister(self, group_id: str):
+        """`on_root_ejected` bound to one group, for that group's indexer."""
+        async def persist(root_name: str, ejected: bool) -> None:
+            if self._roster:
+                await self._roster.set_root_ejected(
+                    group_id, root_name, ejected,
+                    set_by=self._state.get("node_user_id", ""))
+        return persist
+
     async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
         """
         Called when a DirectoryIndexer detects file changes — once per
@@ -1716,10 +1749,10 @@ def main() -> None:
                         help="group id (optional if only one is configured)")
     parser.add_argument("--writable", action="store_true", default=None,
                         dest="writable",
-                        help="mark root as read-write (root set/add)")
+                        help="root accepts member uploads (root add/set)")
     parser.add_argument("--no-writable", action="store_false",
                         dest="writable",
-                        help="mark root as read-only (root set)")
+                        help="root is read-only (root add/set, group add)")
     parser.add_argument("--removable", action="store_true", default=None,
                         dest="removable",
                         help="mark root as removable (root set/add)")
@@ -2481,20 +2514,22 @@ def main() -> None:
             sys.exit(1)
 
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
-        body = {"name": args.target, "shared_dir": args.dir}
+        # Writable unless the operator says otherwise: a brand-new group that
+        # cannot receive a single file until its owner finds a second command
+        # is not a working group. Every root added *later* is read-only by
+        # default, which is the opposite rule and the right one there.
+        writable = args.writable is not False
+        body = {"name": args.target, "shared_dir": args.dir,
+                "writable": writable}
         if args.upload_dir:
-            import warnings
-            warnings.warn(
-                "--upload-dir is deprecated; the main root is writable by "
-                "default. Use 'meshbay-node root add' for additional roots.",
-                DeprecationWarning, stacklevel=1)
             print("WARNING: --upload-dir is deprecated. The shared directory is "
-                  "writable by default. Use 'meshbay-node root add' for "
-                  "additional roots.")
+                  "read-write by default; use 'meshbay-node root add "
+                  "<path> --writable' for a second one.")
             body["upload_dir"] = args.upload_dir
         out = _daemon_api(cfg, "/api/groups/attach", method="POST", body=body)
         print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
-        print(f"  shared_dir   {out['shared_dir']}  (writable)")
+        print(f"  shared_dir   {out['shared_dir']}"
+              f"  ({'read-write' if writable else 'read-only'})")
         if out.get("upload_dir"):
             print(f"  upload_dir   {out['upload_dir']}")
         print()
@@ -2512,7 +2547,7 @@ def main() -> None:
         group_id = _resolve_group(cfg, args.group)
 
         if sub == "list":
-            out = _daemon_api(cfg, f"/api/groups")
+            out = _daemon_api(cfg, "/api/groups")
             group = next((g for g in out.get("groups", [])
                           if g["id"] == group_id), None)
             if not group: