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>

2 files changed, 360 insertions, 50 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
index a69429c..5dbdc5d 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py
@@ -59,6 +59,12 @@ class GroupIndex:
     sk_node:  Ed25519PrivateKey
     gek:      bytes | None = None   # None → public group (no encryption)
     version:  int = 1
+    # The group's roots and whether each is readable right now. Travels inside
+    # the encrypted payload because it names the operator's directories, and a
+    # member needs it to tell "temporarily unavailable" from "deleted" — a
+    # distinction the entries alone cannot carry, since an unavailable root's
+    # files are still listed. Absent in an index written before roots existed.
+    roots:    list = field(default_factory=list)
     _entries: dict = field(default_factory=dict, repr=False)  # id → IndexEntry
 
     # ── Entry management ──────────────────────────────────────────────────────
@@ -90,6 +96,7 @@ class GroupIndex:
         payload = msgpack.packb({
             "group_id": self.group_id,
             "version":  self.version,
+            "roots":    list(self.roots),
             "entries":  [asdict(e) for e in self.entries],
         }, use_bin_type=True)
 
@@ -170,6 +177,9 @@ class GroupIndex:
             sk_node=sk_node,
             gek=gek,
             version=payload["version"],
+            # Absent from an index written before roots existed; an empty list
+            # reads as "nothing known about availability", not "no roots".
+            roots=payload.get("roots") or [],
         )
         for e in payload["entries"]:
             idx.add_entry(IndexEntry(**e))
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index 60dc04b..d9b1d2c 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -1,15 +1,29 @@
 """
-Directory indexer — watches a directory and maintains a GroupIndex.
+Directory indexer — watches a group's roots and maintains a GroupIndex.
 
-Uses watchdog for filesystem events. On any change (create/modify/delete/move),
-the affected file is re-scanned and the GroupIndex is updated.
-File metadata (blake3 hash, size, type, duration) is computed on first scan.
-Heavy operations (hashing large files) run in a thread pool to avoid blocking.
+Uses watchdog for filesystem events. On any change the affected file is
+re-scanned and the GroupIndex is updated. File metadata (blake3 hash, size,
+type) is computed on first scan; hashing runs in a thread pool.
+
+Two properties are worth stating because they are what the code is shaped
+around, not incidental:
+
+**A root that goes away freezes; it never empties.** Unmounting a volume either
+makes watchdog emit a deletion for every file under it or presents an empty
+directory to the next scan. Both would propagate deletions for a whole library
+as though the owner had erased it, to every member. So a deletion is acted on
+only once the root it belongs to has been confirmed still readable, and a root
+that is not is marked unavailable with its entries left exactly where they are.
+
+**Events are not trusted to be complete.** `ReadDirectoryChangesW` drops events
+when its buffer overflows under a burst, and inotify on a FUSE mount misses
+changes made outside it. Most users are on Windows sharing from exFAT, so both
+apply. A periodic reconciliation scan is therefore not a belt-and-braces extra;
+it is the only thing that recovers a missed event.
 """
 
 import asyncio
 import logging
-import mimetypes
 import time
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
@@ -20,8 +34,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from watchdog.events import FileSystemEvent, FileSystemEventHandler
 from watchdog.observers import Observer
 
+from meshbay_common.paths import fold, find_fold_collisions, long_path
 from meshbay_common.protocol import IndexEntry
 from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roots import Root, RootSet
 
 log = logging.getLogger(__name__)
 
@@ -59,7 +75,19 @@ def _is_indexable(path: Path) -> bool:
 _HASH_CHUNK = 8 * 1024 * 1024   # 8 MB streaming hash chunks
 
 
-def _scan_file(root: Path, file_path: Path) -> IndexEntry | None:
+def _virtual_dir(root: Root, file_path: Path) -> str:
+    """
+    The directory a file appears in, as members see it: `"Films/2024"`.
+
+    The root name is the first segment for every root, including the only one of
+    a single-root group — one path shape has to be got right once, two have to
+    be kept right forever.
+    """
+    rel = file_path.parent.relative_to(root.path)
+    return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"
+
+
+def _scan_file(root: Root, file_path: Path) -> IndexEntry | None:
     """Compute IndexEntry for a file. Blocking — run in executor.
     Uses streaming blake3 so arbitrarily large files (ISOs, VM images, etc.)
     don't require loading the whole file into memory."""
@@ -68,33 +96,33 @@ def _scan_file(root: Path, file_path: Path) -> IndexEntry | None:
     try:
         stat = file_path.stat()
         hasher = blake3.blake3()
-        with open(file_path, "rb") as f:
+        # long_path is a no-op off Windows; there it is what lets a deep media
+        # library past MAX_PATH.
+        with open(long_path(file_path), "rb") as f:
             while chunk := f.read(_HASH_CHUNK):
                 hasher.update(chunk)
-        file_id = hasher.hexdigest()
-        rel_path = str(file_path.parent.relative_to(root))
-        if rel_path == ".":
-            rel_path = ""
         return IndexEntry(
-            id=file_id,
+            id=hasher.hexdigest(),
+            # Stored exactly as the filesystem gave it: this is the string that
+            # opens the file. Normalization is for comparison only.
             name=file_path.name,
-            path=rel_path,
+            path=_virtual_dir(root, file_path),
             size=stat.st_size,
             type=_detect_type(file_path),
             added_at=int(stat.st_mtime),
         )
-    except (OSError, PermissionError) as e:
+    except (OSError, PermissionError, ValueError) as e:
         log.warning("Cannot index %s: %s", file_path, e)
         return None
 
 
 class DirectoryIndexer:
     """
-    Watches a directory and keeps a GroupIndex up to date.
+    Watches a group's roots and keeps a GroupIndex up to date.
 
     Usage:
         indexer = DirectoryIndexer(
-            root=Path("/home/user/shared"),
+            roots=RootSet.build([{"path": "/home/user/shared", "upload": True}]),
             group_id="my-group",
             sk_node=sk,
             gek=gek_bytes,
@@ -105,61 +133,147 @@ class DirectoryIndexer:
         await indexer.stop()
     """
 
+    # How often to re-check which roots are readable and reconcile the index
+    # against what is actually on disk. Not a poll for changes — a backstop for
+    # the events the OS did not deliver, and the way a re-plugged drive is
+    # noticed.
+    RECONCILE_SECS = 60.0
+
     def __init__(
         self,
-        root: Path,
+        roots: RootSet,
         group_id: str,
         sk_node: Ed25519PrivateKey,
         gek: bytes | None,
         on_change: Callable[["DirectoryIndexer"], Awaitable[None]] | None = None,
     ):
-        self.root     = root.resolve()
+        self.roots    = roots
         self.group_id = group_id
         self.sk_node  = sk_node
         self.gek      = gek
         self.on_change = on_change
 
         self._index    = GroupIndex(group_id=group_id, sk_node=sk_node, gek=gek)
+        self._index.roots = roots.describe()
         self._executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="indexer")
         self._observer: Observer | None = None
         self._loop: asyncio.AbstractEventLoop | None = None
+        self._reconciler: asyncio.Task | None = None
+        self._pending_timers: dict[str, asyncio.TimerHandle] = {}
 
     @property
     def index(self) -> GroupIndex:
         return self._index
 
+    # ── Root lookup ───────────────────────────────────────────────────────────
+
+    def _root_for(self, file_path: Path) -> Root | None:
+        """Which root a real path belongs to, longest match first."""
+        try:
+            resolved = file_path.resolve()
+        except OSError:
+            resolved = file_path
+        best: Root | None = None
+        for root in self.roots:
+            try:
+                resolved.relative_to(root.path)
+            except ValueError:
+                continue
+            if best is None or len(root.path.parts) > len(best.path.parts):
+                best = root
+        return best
+
     # ── Initial scan ──────────────────────────────────────────────────────────
 
     async def initial_scan(self) -> None:
-        """Scan the entire directory tree. Run once at startup."""
-        log.info("Scanning %s ...", self.root)
+        """Scan every available root. Run once at startup."""
+        self.roots.refresh_availability()
+        total = 0
+        for root in self.roots:
+            if not root.available:
+                log.warning("Root %r is not readable at startup (%s) — its files "
+                            "are not indexed yet and will appear when it returns",
+                            root.name, root.path)
+                continue
+            total += await self._scan_root(root)
+        self._index.version = int(time.time())
+        self._index.roots = self.roots.describe()
+        self._report_collisions()
+        log.info("Initial scan complete: %d files across %d root(s)",
+                 total, len(self.roots))
+
+    async def _scan_root(self, root: Root) -> int:
+        log.info("Scanning %s (root %r) ...", root.path, root.name)
         loop = asyncio.get_event_loop()
-        files = [p for p in self.root.rglob("*") if p.is_file()]
         count = 0
+        try:
+            files = [p for p in root.path.rglob("*") if p.is_file()]
+        except OSError as e:
+            log.warning("Cannot scan root %r: %s", root.name, e)
+            return 0
         for file_path in files:
             entry = await loop.run_in_executor(
-                self._executor, _scan_file, self.root, file_path)
+                self._executor, _scan_file, root, file_path)
             if entry:
                 self._index.add_entry(entry)
                 count += 1
-        self._index.version = int(time.time())
-        log.info("Initial scan complete: %d files indexed", count)
+        return count
+
+    def _report_collisions(self) -> None:
+        """
+        Names that are the same file on a case-insensitive filesystem.
+
+        Reported, never resolved: on ext4 both files exist and only the operator
+        knows which was meant. Left silent, the pair reaches somebody on Windows
+        who can save one of them.
+        """
+        by_dir: dict[str, list[str]] = {}
+        for entry in self._index.entries:
+            by_dir.setdefault(fold(entry.path), []).append(entry.name)
+        for folded_dir, names in by_dir.items():
+            for _, clashing in find_fold_collisions(names).items():
+                log.warning(
+                    "Names that differ only by case or accent form in %s: %s — "
+                    "these are one file on NTFS or exFAT, and a member on Windows "
+                    "can only keep one of them",
+                    folded_dir or "/", ", ".join(sorted(clashing)))
 
     # ── Watchdog integration ──────────────────────────────────────────────────
 
     async def start(self) -> None:
-        """Start initial scan + filesystem watcher."""
+        """Start initial scan + filesystem watcher + reconciler."""
         self._loop = asyncio.get_event_loop()
         await self.initial_scan()
+        self._start_observer()
+        self._reconciler = asyncio.create_task(self._reconcile_loop())
 
+    def _start_observer(self) -> None:
         handler = _WatchdogHandler(self)
         self._observer = Observer()
-        self._observer.schedule(handler, str(self.root), recursive=True)
+        watched = 0
+        for root in self.roots:
+            if not root.available:
+                continue
+            try:
+                self._observer.schedule(handler, str(root.path), recursive=True)
+                watched += 1
+            except OSError as e:
+                log.warning("Cannot watch root %r: %s", root.name, e)
         self._observer.start()
-        log.info("Watching %s for changes", self.root)
+        log.info("Watching %d root(s) for changes", watched)
 
     async def stop(self) -> None:
-        """Stop the filesystem watcher."""
+        """Stop the filesystem watcher and the reconciler."""
+        if self._reconciler:
+            self._reconciler.cancel()
+            try:
+                await self._reconciler
+            except asyncio.CancelledError:
+                pass
+            self._reconciler = None
+        for handle in self._pending_timers.values():
+            handle.cancel()
+        self._pending_timers.clear()
         if self._observer:
             self._observer.stop()
             self._observer.join()
@@ -167,6 +281,171 @@ class DirectoryIndexer:
         self._executor.shutdown(wait=False)
         log.info("Indexer stopped")
 
+    async def retarget(self, roots: RootSet) -> None:
+        """
+        Point this indexer at a new set of roots, without a restart (14.8).
+
+        Entries under a root that is gone from the config are dropped — the
+        operator removed it deliberately, which is not the same event as a
+        volume disappearing, and conflating the two is what §6.9 exists to
+        prevent. Roots that survive keep their entries; new ones are scanned.
+        """
+        old_names = {r.folded for r in self.roots}
+        new_names = {r.folded for r in roots}
+
+        for root in self.roots:
+            if root.folded not in new_names:
+                dropped = self._entries_under(root)
+                log.info("Root %r removed from the config — dropping %d entries",
+                         root.name, len(dropped))
+                for entry in dropped:
+                    self._index.remove_entry(entry.id)
+
+        self.roots = roots
+        roots.refresh_availability()
+        for root in roots:
+            if root.folded not in old_names and root.available:
+                await self._scan_root(root)
+
+        self._index.roots = roots.describe()
+        self._index.version = int(time.time())
+        self._restart_observer()
+        if self.on_change:
+            await self.on_change(self)
+
+    # ── Reconciliation ────────────────────────────────────────────────────────
+
+    async def _reconcile_loop(self) -> None:
+        while True:
+            try:
+                await asyncio.sleep(self.RECONCILE_SECS)
+                await self.reconcile()
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                log.exception("Reconcile failed — continuing")
+
+    async def reconcile(self) -> None:
+        """
+        Re-check availability, and rescan roots that came back.
+
+        The only place a root's entries are dropped: when the root is readable
+        and the files are genuinely gone. A root that is not readable is left
+        untouched, which is the whole point.
+        """
+        changed = self.roots.refresh_availability()
+        touched = False
+
+        for root, available in changed:
+            if available:
+                log.info("Root %r is back — rescanning", root.name)
+                self._drop_root_entries(root)
+                await self._scan_root(root)
+                touched = True
+            else:
+                # Frozen: entries stay, marked unavailable to members through
+                # the roots table in the index payload.
+                log.warning("Root %r went away — %d entries frozen, not deleted",
+                            root.name, len(self._entries_under(root)))
+                touched = True
+
+        if changed:
+            self._restart_observer()
+
+        if await self._sweep_available_roots():
+            touched = True
+
+        if touched:
+            self._index.roots = self.roots.describe()
+            self._index.version = int(time.time())
+            if self.on_change:
+                await self.on_change(self)
+
+    async def _sweep_available_roots(self) -> bool:
+        """
+        Catch what the watcher missed: files gone, and files never announced.
+
+        Only touches roots that are readable right now — a root whose volume is
+        absent has nothing to compare against, and comparing anyway is exactly
+        the mistake this module exists to avoid.
+        """
+        loop = asyncio.get_event_loop()
+        changed = False
+        for root in self.roots:
+            if not root.available:
+                continue
+            try:
+                on_disk = {p.resolve() for p in root.path.rglob("*")
+                           if _is_indexable(p)}
+            except OSError as e:
+                log.warning("Cannot reconcile root %r: %s", root.name, e)
+                continue
+
+            known: dict[Path, str] = {}
+            for entry in self._entries_under(root):
+                abs_path = self._entry_path(root, entry)
+                if abs_path:
+                    known[abs_path] = entry.id
+
+            for missing in set(known) - on_disk:
+                # Duplicate content is handled without a special case here: the
+                # entry goes, and the add loop below re-indexes the surviving
+                # copy under its own path, because the id is then absent. A
+                # dedicated "find the survivor" lookup was written first and
+                # deleted — it rehashed every file under the root on any single
+                # deletion, and a test proved it changed nothing.
+                self._index.remove_entry(known[missing])
+                log.info("Reconcile: %s is gone", missing)
+                changed = True
+
+            for added in on_disk - set(known):
+                entry = await loop.run_in_executor(
+                    self._executor, _scan_file, root, added)
+                if not entry:
+                    continue
+                # The index is keyed by **content hash**, so two identical files
+                # at two paths are one entry and the path comparison above
+                # cannot see the second. Adding it anyway rewrites that entry's
+                # path every cycle, bumps the version, and pushes an index
+                # update to every connected peer once a minute — for ever.
+                # Measured on a live node: `clip.mp4` present at the root and in
+                # uploads/ with the same bytes.
+                if self._index.get_entry(entry.id) is not None:
+                    log.debug("Reconcile: %s duplicates content already indexed "
+                              "as %s — leaving the index alone",
+                              added, entry.id[:8])
+                    continue
+                self._index.add_entry(entry)
+                log.info("Reconcile: %s appeared (missed event)", added)
+                changed = True
+        return changed
+
+    def _entries_under(self, root: Root) -> list[IndexEntry]:
+        prefix = fold(root.name)
+        return [e for e in self._index.entries
+                if fold(e.path).split("/", 1)[0] == prefix]
+
+    def _drop_root_entries(self, root: Root) -> None:
+        for entry in self._entries_under(root):
+            self._index.remove_entry(entry.id)
+
+    @staticmethod
+    def _entry_path(root: Root, entry: IndexEntry) -> Path | None:
+        _, _, tail = entry.path.partition("/")
+        try:
+            return (root.path / tail / entry.name).resolve() if tail else \
+                   (root.path / entry.name).resolve()
+        except OSError:
+            return None
+
+    def _restart_observer(self) -> None:
+        """Re-schedule watches after roots appeared or disappeared."""
+        if self._observer:
+            self._observer.stop()
+            self._observer.join()
+            self._observer = None
+        self._start_observer()
+
     # ── Internal update ───────────────────────────────────────────────────────
 
     _DEBOUNCE_SECS = 2.0
@@ -175,39 +454,60 @@ class DirectoryIndexer:
         """Called from watchdog thread — schedule debounced async update."""
         if not self._loop:
             return
-        key = str(file_path.resolve())
-        self._loop.call_soon_threadsafe(
-            self._debounce, key, file_path, deleted)
+        key = str(file_path)
+        self._loop.call_soon_threadsafe(self._debounce, key, file_path, deleted)
 
     def _debounce(self, key: str, file_path: Path, deleted: bool) -> None:
-        if not hasattr(self, "_pending_timers"):
-            self._pending_timers: dict[str, asyncio.TimerHandle] = {}
         old = self._pending_timers.pop(key, None)
         if old:
             old.cancel()
-        handle = self._loop.call_later(
-            self._DEBOUNCE_SECS,
-            lambda: asyncio.ensure_future(self._update_entry(file_path, deleted)),
-        )
-        self._pending_timers[key] = handle
 
-    def _remove_by_path(self, file_path: Path) -> None:
-        """Remove any existing entries that match this file's path + name."""
-        resolved = file_path.resolve()
-        to_remove = [
-            e.id for e in self._index.entries
-            if (self.root / e.path / e.name).resolve() == resolved
-        ]
-        for fid in to_remove:
-            self._index.remove_entry(fid)
+        def fire() -> None:
+            self._pending_timers.pop(key, None)
+            asyncio.ensure_future(self._update_entry(file_path, deleted))
+
+        self._pending_timers[key] = self._loop.call_later(self._DEBOUNCE_SECS, fire)
+
+    def _remove_by_path(self, root: Root, file_path: Path) -> None:
+        """Remove any existing entries that point at this file."""
+        try:
+            resolved = file_path.resolve()
+        except OSError:
+            resolved = file_path
+        for entry in self._entries_under(root):
+            if self._entry_path(root, entry) == resolved:
+                self._index.remove_entry(entry.id)
 
     async def _update_entry(self, file_path: Path, deleted: bool) -> None:
-        self._remove_by_path(file_path)
+        root = self._root_for(file_path)
+        if root is None:
+            return
+
+        if deleted and not root.is_live():
+            # The volume went away rather than the file. Freeze: mark the root
+            # and touch nothing. Every other event for this root will arrive
+            # here too and be dropped the same way, which is the intent — one
+            # unplugged drive must not empty a library.
+            if root.available:
+                root.available = False
+                self._index.roots = self.roots.describe()
+                log.warning("Root %r disappeared — ignoring deletion events and "
+                            "freezing %d entries", root.name,
+                            len(self._entries_under(root)))
+                self._index.version = int(time.time())
+                if self.on_change:
+                    await self.on_change(self)
+            return
+
+        if not root.available:
+            return
+
+        self._remove_by_path(root, file_path)
 
         if not deleted:
             loop = asyncio.get_event_loop()
             entry = await loop.run_in_executor(
-                self._executor, _scan_file, self.root, file_path)
+                self._executor, _scan_file, root, file_path)
             if entry:
                 self._index.add_entry(entry)
                 log.debug("Indexed: %s (%s, %d bytes)",