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, 322 insertions, 0 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py
new file mode 100644
index 0000000..8f999d7
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/roots.py
@@ -0,0 +1,322 @@
+"""
+A group's content is several named roots, not one directory.
+
+    /                       (the group's virtual root)
+    ├── Films/          →   D:\\Media\\Films
+    ├── Musique/        →   E:\\Audio            (external drive)
+    └── Documents/      →   C:\\Users\\me\\Partage
+
+Every index path carries the root name as its first segment, uniformly — a group
+with one root is not a special case, because two path shapes would have to be
+kept right forever and one shape only has to be got right once.
+
+Three rules, and each of them is load-bearing rather than tidy:
+
+  * **The name is the directory's basename, derived once and stored.** Never
+    recomputed from the path, or renaming a folder on disk silently re-identifies
+    every file under it.
+  * **No root may contain another.** Otherwise the same bytes are indexed twice
+    under two identities, and deleting one leaves the other pointing at nothing.
+  * **Availability is per root.** A root whose volume goes away *freezes*: its
+    entries stay in the index, marked unavailable. Emptying it would propagate
+    deletions for a whole library as though the owner had erased it.
+
+Comparison of names is case-insensitive and NFC-normalized (`meshbay_common.paths`),
+because most of these directories live on exFAT or NTFS, where `Films` and `films`
+are one directory.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from meshbay_common.paths import fold, portable_name_problem
+
+log = logging.getLogger(__name__)
+
+VALID_KINDS = ("generic", "video", "audio", "photo")
+
+
+class RootError(ValueError):
+    """A root set that cannot be built. The message is shown to the operator."""
+
+
+@dataclass
+class Root:
+    """One named directory inside a group."""
+
+    name: str
+    path: Path
+    kind: str = "generic"
+    upload: bool = False
+    # Runtime, not configuration: set by the indexer when the directory can no
+    # longer be read, and cleared when it comes back.
+    available: bool = True
+
+    @property
+    def folded(self) -> str:
+        return fold(self.name)
+
+    def is_live(self) -> bool:
+        """Readable right now. The question `available` caches."""
+        try:
+            return self.path.is_dir()
+        except OSError:
+            return False
+
+
+def derive_name(path: Path) -> str:
+    """
+    The name a directory gets when it is added: its basename.
+
+    A path that has no usable basename — a drive root such as `E:\\`, or `/` —
+    has nothing to derive from, and the operator has to supply a name.
+    """
+    name = path.name or ""
+    if not name:
+        raise RootError(
+            f"{path} has no directory name to use — give the root an explicit "
+            f"name (a drive or filesystem root cannot supply one)")
+    return name
+
+
+@dataclass
+class RootSet:
+    """
+    The roots of one group, and the only place a virtual path is resolved.
+
+    `resolve()` is the single entry point for turning something that arrived
+    over the wire into a path on disk. Callers must not join paths themselves —
+    that is how a traversal gets in through the one site nobody reviewed.
+    """
+
+    roots: list[Root] = field(default_factory=list)
+
+    # ── Construction ─────────────────────────────────────────────────────────
+
+    @classmethod
+    def build(cls, specs: list[dict]) -> "RootSet":
+        """
+        Build from configuration, refusing anything ambiguous.
+
+        `specs` are dicts with `path`, and optionally `name`, `kind`, `upload`.
+        Raises RootError with a message meant for an operator reading a log.
+        """
+        roots: list[Root] = []
+        by_folded: dict[str, Root] = {}
+
+        for spec in specs:
+            raw = str(spec.get("path", "")).strip()
+            if not raw:
+                raise RootError("a root has no path")
+            path = Path(raw).expanduser()
+            try:
+                path = path.resolve()
+            except OSError as e:
+                raise RootError(f"{raw}: {e}") from e
+
+            name = str(spec.get("name") or "").strip() or derive_name(path)
+
+            problem = portable_name_problem(name)
+            if problem:
+                raise RootError(
+                    f"root name {name!r} ({problem}) — every member sees this as a "
+                    f"folder name, including on Windows. Give the root an explicit "
+                    f"name in the config")
+
+            clash = by_folded.get(fold(name))
+            if clash:
+                raise RootError(
+                    f"two roots would both be called {name!r}: {clash.path} and "
+                    f"{path}. Names are compared without regard to case. Give one "
+                    f"of them an explicit name")
+
+            kind = str(spec.get("kind") or "generic").strip().lower()
+            if kind not in VALID_KINDS:
+                log.warning("root %r: unknown kind %r — using 'generic'", name, kind)
+                kind = "generic"
+
+            root = Root(name=name, path=path, kind=kind,
+                        upload=bool(spec.get("upload", False)))
+            _refuse_nesting(root, roots)
+            roots.append(root)
+            by_folded[root.folded] = root
+
+        cls._settle_upload_root(roots)
+        return cls(roots=roots)
+
+    @staticmethod
+    def _settle_upload_root(roots: list[Root]) -> None:
+        """
+        Exactly one root receives uploads, and the operator picks it.
+
+        Not guessed when several are marked, because "uploads went somewhere
+        else" is discovered weeks later. With none marked and a single root, the
+        answer is not ambiguous, so it is taken.
+        """
+        marked = [r for r in roots if r.upload]
+        if len(marked) > 1:
+            names = ", ".join(r.name for r in marked)
+            raise RootError(
+                f"several roots are marked upload = true ({names}) — exactly one "
+                f"receives uploads")
+        if not marked and len(roots) == 1:
+            roots[0].upload = True
+
+    # ── Lookup ───────────────────────────────────────────────────────────────
+
+    def by_name(self, name: str) -> Root | None:
+        target = fold(name)
+        for root in self.roots:
+            if root.folded == target:
+                return root
+        return None
+
+    @property
+    def upload_root(self) -> Root | None:
+        for root in self.roots:
+            if root.upload:
+                return root
+        return None
+
+    @property
+    def names(self) -> list[str]:
+        return [r.name for r in self.roots]
+
+    def __bool__(self) -> bool:
+        return bool(self.roots)
+
+    def __len__(self) -> int:
+        return len(self.roots)
+
+    def __iter__(self):
+        return iter(self.roots)
+
+    # ── Resolution ───────────────────────────────────────────────────────────
+
+    def split(self, virtual: str) -> tuple[Root, str] | None:
+        """
+        `"Films/2024"` → (the Films root, `"2024"`). None if no such root.
+
+        Does not touch the filesystem, so it is safe to call on a root whose
+        volume is absent.
+        """
+        rel = (virtual or "").strip().strip("/")
+        if not rel:
+            return None
+        head, _, tail = rel.partition("/")
+        root = self.by_name(head)
+        if root is None:
+            return None
+        return root, tail
+
+    def resolve(self, virtual: str, *, require_available: bool = True) -> Path | None:
+        """
+        A virtual path from the wire → a real path inside its root, or None.
+
+        Refuses `..`, absolute segments, and anything whose resolved form escapes
+        the root — symlinks included, which is why this resolves before
+        comparing rather than checking the string.
+        """
+        found = self.split(virtual)
+        if found is None:
+            return None
+        root, tail = found
+        if require_available and not root.available:
+            return None
+
+        parts = [seg for seg in tail.split("/") if seg not in ("", ".")]
+        if any(seg == ".." for seg in parts):
+            return None
+        try:
+            target = (root.path / Path(*parts)).resolve() if parts else root.path.resolve()
+            base = root.path.resolve()
+        except OSError:
+            return None
+        if target != base and base not in target.parents:
+            return None
+        return target
+
+    def virtual_of(self, absolute: Path) -> str | None:
+        """A real path anywhere in this group → `"Films/2024"`, or None."""
+        for root in self.roots:
+            virtual = self.virtual_path(root, absolute)
+            if virtual is not None:
+                return virtual
+        return None
+
+    def virtual_path(self, root: Root, absolute: Path) -> str | None:
+        """The inverse of `resolve`: a real path → `"Films/2024"`."""
+        try:
+            rel = absolute.resolve().relative_to(root.path.resolve())
+        except (ValueError, OSError):
+            return None
+        return root.name if str(rel) == "." else f"{root.name}/{rel.as_posix()}"
+
+    # ── Availability ─────────────────────────────────────────────────────────
+
+    def refresh_availability(self) -> list[tuple[Root, bool]]:
+        """
+        Re-read which roots are readable. Returns the ones that changed.
+
+        Called periodically and after a filesystem event that looks like a
+        disappearance. A change here never edits the index: a root going away
+        freezes its entries, and a root coming back triggers a rescan.
+        """
+        changed: list[tuple[Root, bool]] = []
+        for root in self.roots:
+            live = root.is_live()
+            if live != root.available:
+                root.available = live
+                changed.append((root, live))
+                log.warning("Root %r is now %s (%s)", root.name,
+                            "available" if live else "UNAVAILABLE", root.path)
+        return changed
+
+    def describe(self) -> list[dict]:
+        """Per-root state for the index payload and the admin UI."""
+        return [
+            {"name": r.name, "kind": r.kind, "available": r.available,
+             "upload": r.upload}
+            for r in self.roots
+        ]
+
+
+def entry_abs_path(roots: RootSet, entry) -> Path | None:
+    """
+    Where an index entry actually lives, or None if its root is gone.
+
+    The one place an entry becomes a path. `entry.path` starts with a root name,
+    so a group whose drive is unplugged answers None here rather than opening
+    something unrelated that happens to share a relative path with another root.
+    """
+    parent = roots.resolve(entry.path)
+    return (parent / entry.name) if parent else None
+
+
+def _refuse_nesting(new: Root, existing: list[Root]) -> None:
+    """
+    No root may contain another, compared case-insensitively.
+
+    `D:\\Media` and `D:\\Media\\Films` together would index the same bytes twice
+    under two identities. On NTFS and exFAT `d:\\media` is the same directory as
+    `D:\\Media`, so a string comparison that respects case would miss it.
+    """
+    new_parts = [fold(p) for p in new.path.parts]
+    for other in existing:
+        other_parts = [fold(p) for p in other.path.parts]
+        if new_parts == other_parts:
+            raise RootError(
+                f"roots {new.name!r} and {other.name!r} are the same directory "
+                f"({new.path})")
+        shorter, longer, inner, outer = (
+            (other_parts, new_parts, new, other)
+            if len(other_parts) < len(new_parts)
+            else (new_parts, other_parts, other, new))
+        if longer[:len(shorter)] == shorter:
+            raise RootError(
+                f"root {inner.name!r} ({inner.path}) is inside root "
+                f"{outer.name!r} ({outer.path}) — its files would be indexed "
+                f"twice under two names")