feat(node): several named roots per group, and one implementation per operation

Stage A — a group's content is a set of named roots
---------------------------------------------------
`shared_dir` becomes a list of {name, path, kind}. The name is the directory's
basename, derived once at add time and *stored*: recomputing it would
re-identify a whole library the day someone renames a folder on disk. Duplicate
names are refused case-insensitively and no root may contain another — both
compared with NFC folding, because most of these directories live on exFAT or
NTFS where `Films` and `films` are one directory.

Every index path carries its root name, in a one-root group as much as in a
five-root one. One path shape has to be got right once; two have to be kept
right for ever.

**A root that goes away freezes; it never empties.** Unmounting a volume makes
watchdog report every file under it as deleted, or presents an empty directory
to the next scan. Acting on either propagates deletions for a whole library to
every member, as though the owner had erased it. So a deletion is acted on only
once its root is confirmed readable, and availability is tracked per root — one
unplugged drive leaves the others serving. 12 tests, verified to fail against an
indexer without the check.

Events are not trusted to be complete either: ReadDirectoryChangesW drops them
under load and inotify on a FUSE mount misses changes made outside it. A
periodic reconciliation sweep is the only thing that recovers a missed event.

MNP 0.2 → 0.3 (additive). The hub needs no change: SwarmSource carries a content
hash, a node id and an endpoint — no paths, no filenames — and private groups
register nothing (H7).

Stage B — one implementation behind every front door
----------------------------------------------------
C1 and C6 were both "a second path into the node with its own weaker
handshake". Two implementations of `revoke` with two authorization checks is
that shape one size down. `meshbay_node/ops.py` holds each operation once,
takes the daemon state, and knows nothing about HTTP, argv or MNP. The loopback
API is one `_op(...)` line per endpoint; the MNP handlers call the same
functions. test_ops.py asserts the shape rather than trusting it.

Phase 14 is finished on top of it — `group list`, `gek init|rotate`, `reload`
(SIGHUP), `denylist show|clear`, `file list|rm`. **No operator action requires a
browser any more.** Plus `gek_rotate` and `member_unpin` as operator-signed MNP
operations: rotation is the half of revocation that revocation cannot do, since
the ex-member holds the current key, and the node generates the replacement
with its own CSPRNG — no key material crosses the wire, which is what the C5b
rule is actually about.

Two bugs found by running it rather than by testing it
------------------------------------------------------
GroupIndex is keyed by **content hash**, so the same bytes at two paths are one
entry — which is also why a scan reports ten files and indexes nine.
Reconciliation compared paths, so it decided the second path was a missed event
every 60 s, rewrote the entry and pushed an index update to every connected
peer. Seen in a live node's log.

`meshbay-node reload` crashed on first use with `subprocess` unimported: the
module compiles fine, which is the "syntax, not names" trap already recorded for
the SPA. test_cli_dispatch.py now walks every verb and refuses to let one be
added to the parser without an entry there.

Also corrected: protocol.py declared a second MNP_VERSION of "0.1" while the
wire carried "0.2" — harmless only because nothing imported it. And
_do_dir_create/_do_dir_delete referenced an undefined `filename` on their error
path.

740 tests pass; QE/deploy/e2e.py passes end to end against the live deployment.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

1 files changed, 288 insertions, 29 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 8b4a1d7..21331f2 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -26,6 +26,7 @@ Usage:
 """
 
 import asyncio
+from dataclasses import asdict
 import base64
 import json
 import logging
@@ -42,6 +43,7 @@ from meshbay_node.audit import AuditStore
 from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
 from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
+from meshbay_node.roots import RootSet, RootError
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.keystore import load_or_create_keystore
@@ -197,17 +199,30 @@ class NodeDaemon:
             # 4. Build per-group contexts
             groups_ctx: dict[str, dict] = {}
             for group_cfg in self._config.groups:
-                if not group_cfg.id or not group_cfg.shared_dir:
-                    log.warning("Group %r missing id or shared_dir — skipping",
-                                group_cfg.name)
+                if not group_cfg.id or not group_cfg.roots:
+                    log.warning("Group %r has no id or no shared directory — "
+                                "skipping", group_cfg.name)
                     continue
 
-                shared_root = Path(group_cfg.shared_dir).expanduser().resolve()
-                if not shared_root.exists():
-                    log.warning("Shared dir not found: %s — skipping group %s",
-                                shared_root, group_cfg.name)
+                try:
+                    roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+                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.
+                    log.error("Group %r: %s — skipping", group_cfg.name, e)
                     continue
 
+                roots.refresh_availability()
+                if not any(r.available for r in roots):
+                    # Not skipped for being empty: a group whose only drive is
+                    # unplugged still exists, and its index is frozen rather
+                    # than lost. But there is nothing to serve until it returns.
+                    log.warning(
+                        "Group %r: none of its %d root(s) are readable right now "
+                        "(%s) — serving nothing until one returns",
+                        group_cfg.name, len(roots),
+                        ", ".join(str(r.path) for r in roots))
+
                 gek = None
                 if group_cfg.visibility == "private":
                     gek = await self._load_gek(
@@ -219,7 +234,7 @@ class NodeDaemon:
                                  group_cfg.name)
 
                 indexer = DirectoryIndexer(
-                    root=shared_root,
+                    roots=roots,
                     group_id=group_cfg.id,
                     sk_node=keys.sk_ed25519,
                     gek=gek,
@@ -229,11 +244,13 @@ class NodeDaemon:
                 self._indexers.append(indexer)
                 self._state["indexes"][group_cfg.id] = indexer.index
                 log.info("Indexing group %s: %s (%d files)",
-                         group_cfg.name, shared_root, indexer.index.count)
+                         group_cfg.name,
+                         ", ".join(f"{r.name}={r.path}" for r in roots),
+                         indexer.index.count)
 
                 groups_ctx[group_cfg.id] = {
                     "gek": gek,
-                    "shared_root": shared_root,
+                    "roots": roots,
                     "index": indexer.index,
                     "visibility": group_cfg.visibility,
                     # Admission policy comes from node.toml, never from the hub:
@@ -270,7 +287,7 @@ class NodeDaemon:
                     sk_node=keys.sk_ed25519,
                     hub_pk_pem=session.hub_pk_pem,
                     gek=first["gek"],
-                    shared_root=first["shared_root"],
+                    roots=first["roots"],
                     index=first["index"],
                     groups=groups_ctx,
                     denylist=denylist,
@@ -290,6 +307,11 @@ class NodeDaemon:
                 self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64
 
                 self._webrtc._ctx["roster"] = self._roster
+                # The MNP adapter calls the same operations as the loopback API
+                # (meshbay_node.ops), and those take the daemon's state. Handing
+                # the transport a second set of lookups is how two paths to one
+                # operation start disagreeing — the shape of C1 and C6.
+                self._webrtc._ctx["daemon_state"] = self._state
                 self._webrtc._ctx["invite_ttl"] = (
                     self._config.node.invite_ttl_hours * 3600)
                 paired = await self._roster.has_operator() if self._roster else False
@@ -310,7 +332,7 @@ class NodeDaemon:
                     sk_node=keys.sk_ed25519,
                     hub_pk_pem=session.hub_pk_pem,
                     gek=first["gek"],
-                    shared_root=first["shared_root"],
+                    roots=first["roots"],
                     index=first["index"],
                     host="::",
                     port=self._config.node.quic_port,
@@ -383,7 +405,13 @@ class NodeDaemon:
             self._state["roster"] = self._roster
             self._state["node_user_id"] = session.user_id
             self._state["webrtc"] = self._webrtc
+            self._state["quic_server"] = self._quic_server
             self._state["hub"] = hub
+            # Rotating a key has to reach every transport holding a copy of it,
+            # and clearing the denylist has to reach the one the handshake
+            # consults — so both are published rather than reachable only
+            # through the object that happens to own them.
+            self._state["denylist"] = self._denylist
             self._state["pk_x25519_raw"] = pk_x_raw
 
             self._state["status"] = "running"
@@ -410,10 +438,78 @@ class NodeDaemon:
             loop = asyncio.get_event_loop()
             for sig in (signal.SIGINT, signal.SIGTERM):
                 loop.add_signal_handler(sig, stop_event.set)
+            # Milestone 14.8: re-read node.toml without dropping connections.
+            try:
+                loop.add_signal_handler(
+                    signal.SIGHUP,
+                    lambda: asyncio.ensure_future(self._reload_config()))
+            except (NotImplementedError, AttributeError):
+                pass          # no SIGHUP on Windows; `reload` says so there
             await stop_event.wait()
 
         await self._shutdown()
 
+    async def _reload_config(self) -> None:
+        """
+        Re-read node.toml on SIGHUP.
+
+        Deliberately narrow: it picks up **root changes on groups already
+        hosted**, which is what an operator adjusts day to day, and reports
+        anything else as needing a restart. Adding or removing a whole group
+        means new indexers, chat stores, GEK loads and transport contexts, and
+        doing that under a live daemon is how a half-built group ends up serving
+        content. Saying "restart for that" is honest and costs one restart.
+
+        Nothing here touches connections: a member watching a film keeps
+        watching it.
+        """
+        log.info("SIGHUP — re-reading %s", self._config_path)
+        try:
+            fresh = load_config(self._config_path)
+        except Exception as e:
+            log.error("Reload failed, keeping the running config: %s", e)
+            return
+
+        groups_ctx = self._state.get("groups_ctx") or {}
+        hosted = set(groups_ctx)
+        incoming = {g.id for g in fresh.groups if g.id}
+        if incoming != hosted:
+            added = ", ".join(sorted(incoming - hosted)) or "none"
+            removed = ", ".join(sorted(hosted - incoming)) or "none"
+            log.warning("Group set changed (added: %s, removed: %s) — restart the "
+                        "daemon for that; roots of existing groups reloaded anyway",
+                        added, removed)
+
+        changed = 0
+        for group_cfg in fresh.groups:
+            ctx = groups_ctx.get(group_cfg.id)
+            if not ctx:
+                continue
+            try:
+                roots = RootSet.build([asdict(r) for r in group_cfg.roots])
+            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:
+                continue
+            roots.refresh_availability()
+            indexer = next((i for i in self._indexers
+                            if i.group_id == group_cfg.id), None)
+            if indexer is None:
+                continue
+            log.info("Group %r roots changed: %s", group_cfg.name,
+                     ", ".join(f"{r.name}={r.path}" for r in roots))
+            await indexer.retarget(roots)
+            ctx["roots"] = roots
+            changed += 1
+
+        self._config = fresh
+        self._state["config"] = fresh
+        log.info("Reload complete — %d group(s) re-rooted", changed)
+
     async def _login_with_retry(self, hub: HubClient):
         """Login to hub, retrying if the node key hasn't been linked yet."""
         import httpx as _httpx
@@ -675,21 +771,27 @@ def main() -> None:
 
     parser = argparse.ArgumentParser(description="MeshBay Node daemon")
     parser.add_argument("command", nargs="?",
-                        choices=["init", "status", "ui", "gek-init", "operator",
-                                 "member", "group", "calibrate-argon2"],
+                        choices=["init", "status", "ui", "gek-init", "gek",
+                                 "operator", "member", "group", "file",
+                                 "denylist", "reload", "calibrate-argon2"],
                         help="init: write example config | status: node state and keys "
                              "| ui: print the admin UI URL | operator pair: pair a "
                              "browser with this node | member list|invite|revoke|unpin "
-                             "| group add <name> --dir <path>: host another of your "
-                             "groups | calibrate-argon2: benchmark")
+                             "| group list|add | gek init|rotate | file list|rm "
+                             "| denylist show|clear | reload: re-read node.toml "
+                             "| calibrate-argon2: benchmark")
     parser.add_argument("subcommand", nargs="?",
                         help="'pair' for operator; list|invite|revoke|unpin for "
-                             "member; 'add' for group")
+                             "member; list|add for group; init|rotate for gek; "
+                             "list|rm for file; show|clear for denylist")
     parser.add_argument("target", nargs="?",
-                        help="username for member invite|revoke|unpin; "
-                             "group name for group add")
+                        help="username for member invite|revoke|unpin; group name "
+                             "for group add; file id for file rm; identifier for "
+                             "denylist clear")
     parser.add_argument("--dir", default=None,
                         help="shared directory, for group add")
+    parser.add_argument("--yes", action="store_true",
+                        help="skip the confirmation for destructive commands")
     parser.add_argument("--config", type=Path, default=None,
                         help="Config file path")
     parser.add_argument("--group", default=None,
@@ -699,8 +801,8 @@ def main() -> None:
     args = parser.parse_args()
 
     # Query commands print a report; library logging would interleave with it.
-    quiet = args.command in ("status", "ui", "gek-init", "operator", "member",
-                             "group")
+    quiet = args.command in ("status", "ui", "gek-init", "gek", "operator",
+                             "member", "group", "file", "denylist", "reload")
     logging.basicConfig(
         level=logging.ERROR if quiet else getattr(logging, args.log_level),
         format="%(asctime)s %(levelname)-8s %(name)s: %(message)s",
@@ -755,11 +857,17 @@ def main() -> None:
         print(f"config       {DEFAULT_CONFIG_PATH}")
         if not cfg.groups:
             print("groups       none configured — create a group on the hub, then add")
-            print("             a [[groups]] entry with its id and shared_dir")
+            print("             a [[groups]] entry with its id and a directory")
         else:
             for g in cfg.groups:
                 print(f"  group      {g.name} [{g.visibility}] {g.id or '<no id>'}")
-                print(f"             {g.shared_dir or '<no shared_dir>'}")
+                if not g.roots:
+                    print("             <no directory configured>")
+                for r in g.roots:
+                    label = r.name or Path(r.path).name
+                    flag = "  (uploads)" if r.upload else ""
+                    live = "" if Path(r.path).expanduser().is_dir() else "  [UNAVAILABLE]"
+                    print(f"             {label} → {r.path}{flag}{live}")
         # Node authority: the roster is the source of truth, node.toml the legacy
         # form. Read the DB directly so this reports correctly while the daemon is
         # stopped — the state an operator is most often in when checking.
@@ -874,22 +982,172 @@ def main() -> None:
         print("usage: meshbay-node member list|invite|revoke|unpin")
         sys.exit(1)
 
-    if args.command == "gek-init":
+    if args.command in ("gek-init", "gek"):
+        # `gek-init` is the original spelling and still works. `gek rotate` is
+        # the one that matters after a revocation: the ex-member holds the
+        # current key and nothing else takes it from them.
+        sub = "init" if args.command == "gek-init" else (args.subcommand or "init")
+        if sub not in ("init", "rotate"):
+            print("usage: meshbay-node gek init|rotate [--group NAME]")
+            sys.exit(1)
+
         cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
         group_id = _resolve_group(cfg, args.group)
-        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek",
+
+        if sub == "rotate" and not args.yes:
+            print("Rotating replaces this group's key.")
+            print("  · every member re-receives it automatically on their next connect")
+            print("  · anyone revoked keeps the OLD key and loses access to new content")
+            print("  · content already downloaded stays readable to whoever has it")
+            if input("rotate now? [y/N] ").strip().lower() not in ("y", "yes"):
+                print("cancelled")
+                return
+
+        out = _daemon_api(cfg, f"/api/groups/{group_id}/gek"
+                               f"{'?rotate=true' if sub == 'rotate' else ''}",
                           method="POST", timeout=60)
 
-        print(f"GEK ready for {group_id}")
+        verb = "rotated" if out.get("rotated") else "ready"
+        print(f"GEK {verb} for {group_id}")
         print(f"  {out.get('authorized_members', 0)} authorized member(s) — each "
               f"receives the key on connect")
         for err in out.get("errors") or []:
             print(f"  ! {err}")
         return
 
+    if args.command == "reload":
+        # Milestone 14.8. The daemon re-reads node.toml; groups that appeared or
+        # whose roots changed are picked up without dropping live connections.
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        import os as _os
+        import signal as _signal
+        import subprocess as _subprocess
+        # `--` is pgrep's own end-of-options marker and must be its own argument;
+        # folded into the pattern it searches for a process literally called
+        # "-- -m …". Anchored to the end of the command line so it matches the
+        # daemon and never a shell that merely mentions it — the same trap
+        # deploy-node.sh documents, where an unanchored pattern kills the script.
+        pid_out = _subprocess.run(
+            ["pgrep", "-f", "--", r"-m meshbay_node\.daemon$"],
+            capture_output=True, text=True)
+        pids = [int(x) for x in pid_out.stdout.split()]
+        if not pids:
+            print("Node is not running — start it with: meshbay-node")
+            sys.exit(1)
+        for pid in pids:
+            _os.kill(pid, _signal.SIGHUP)
+        print(f"sent SIGHUP to {len(pids)} daemon process(es)")
+        print("watch the result:  tail -f /tmp/meshbay-node.log")
+        return
+
+    if args.command == "denylist":
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        sub = args.subcommand or "show"
+
+        if sub == "show":
+            out = _daemon_api(cfg, "/api/denylist")
+            total = out.get("count", 0)
+            if not total:
+                print("denylist     empty — nothing is being refused")
+                return
+            for kind in ("users", "groups", "jtis"):
+                for entry in out.get(kind, []):
+                    print(f"  {kind[:-1]:<6} {entry}")
+            print(f"\n{total} entr(y/ies). These survive a restart (finding H4).")
+            return
+
+        if sub == "clear":
+            if not args.yes:
+                what = args.target or "EVERY entry"
+                print(f"Clearing the denylist re-admits {what}.")
+                print("A revocation the hub sent will not come back on its own.")
+                if input("clear now? [y/N] ").strip().lower() not in ("y", "yes"):
+                    print("cancelled")
+                    return
+            out = _daemon_api(cfg, f"/api/denylist/clear?subject={args.target or ''}",
+                              method="POST")
+            print(f"removed {out['removed']} entr(y/ies) ({out['subject']})")
+            return
+
+        print("usage: meshbay-node denylist show|clear [identifier] [--yes]")
+        sys.exit(1)
+
+    if args.command == "file":
+        cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+        sub = args.subcommand or "list"
+        group_id = _resolve_group(cfg, args.group)
+
+        if sub == "list":
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/files")
+            files = sorted(out.get("files", []), key=lambda f: (f["path"], f["name"]))
+            if not files:
+                print("no files indexed")
+                return
+            for f in files:
+                print(f"  {f['id'][:12]}  {f['size']:>12}  {f['path']}/{f['name']}")
+            print(f"\n{len(files)} file(s). Remove one with: "
+                  f"meshbay-node file rm <id>")
+            return
+
+        if sub == "rm":
+            # Milestone 14.11 — the last operator action that needed a browser.
+            if not args.target:
+                print("usage: meshbay-node file rm <file-id> [--group NAME]")
+                sys.exit(1)
+            out = _daemon_api(cfg, f"/api/groups/{group_id}/files",)
+            matches = [f for f in out.get("files", [])
+                       if f["id"].startswith(args.target)]
+            if not matches:
+                print(f"no file whose id starts with {args.target!r}")
+                sys.exit(1)
+            if len(matches) > 1:
+                print(f"{args.target!r} matches {len(matches)} files — be more specific:")
+                for f in matches[:10]:
+                    print(f"  {f['id'][:16]}  {f['path']}/{f['name']}")
+                sys.exit(1)
+            target = matches[0]
+            if not args.yes:
+                print(f"Delete {target['path']}/{target['name']} "
+                      f"({target['size']} bytes) from disk?")
+                print("This removes the file itself, not just the listing.")
+                if input("delete? [y/N] ").strip().lower() not in ("y", "yes"):
+                    print("cancelled")
+                    return
+            _daemon_api(cfg, f"/api/groups/{group_id}/files/{target['id']}",
+                        method="DELETE")
+            print(f"deleted {target['path']}/{target['name']}")
+            return
+
+        print("usage: meshbay-node file list|rm <id> [--group NAME] [--yes]")
+        sys.exit(1)
+
     if args.command == "group":
+        if args.subcommand in (None, "list"):
+            # Milestone 14.2.
+            cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
+            out = _daemon_api(cfg, "/api/groups")
+            groups = out.get("groups", [])
+            if not groups:
+                print("no groups hosted — add one with: "
+                      "meshbay-node group add <name> --dir <path>")
+                return
+            for g in groups:
+                key = "GEK" if g.get("has_gek") else "NO KEY"
+                print(f"  {g['name']}  [{g['visibility']}/{g.get('join_policy')}]  "
+                      f"{key}  {g['file_count']} file(s)  "
+                      f"{g.get('peers', 0)} peer(s)")
+                print(f"      {g['id']}")
+                for r in g.get("roots", []):
+                    flags = " (uploads)" if r.get("upload") else ""
+                    live = "" if r.get("available", True) else "  [UNAVAILABLE]"
+                    print(f"      root {r['name']}{flags}{live}")
+                if not g.get("has_gek"):
+                    print(f"      give it a key:  meshbay-node gek init "
+                          f"--group {g['name']}")
+            return
+
         if args.subcommand != "add":
-            print("usage: meshbay-node group add <name> --dir <path>")
+            print("usage: meshbay-node group list|add <name> --dir <path>")
             sys.exit(1)
         if not args.target or not args.dir:
             print("usage: meshbay-node group add <name> --dir <path>")
@@ -904,8 +1162,9 @@ def main() -> None:
         print(f"{out['name']} ({out['group_id'][:8]}) added to {out['config']}")
         print(f"  shared_dir   {out['shared_dir']}")
         print()
-        print("Restart the daemon so it picks the group up, then give it a key:")
-        print(f"  meshbay-node gek-init --group {out['name']}")
+        print("Tell the daemon to re-read its config, then give the group a key:")
+        print("  meshbay-node reload")
+        print(f"  meshbay-node gek init --group {out['name']}")
         print()
         print("The key is this group's own — members of your other groups cannot")
         print("read it, and joining one says nothing about the other.")