feat: let the operator pause and resume a root's indexing (MNP 6.2)

A whole-root walk now checks its root before every file: removed or
ejected, it stops instead of reading the drive to the end; paused, it
finishes from the hash cache alone. A paused root is not watched, not
swept by reconcile and survives a restart (roster), its indexed files
stay served. Resume runs the reconcile sweep over the root, so entries
and their enrichment are kept.

Signed root_pause / root_resume over MNP, loopback routes, CLI
`root pause|resume`, a pause button on the indexing dock and
pause/resume in the roots table of Group settings.

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

1 files changed, 72 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/ops/roots.py b/packages/meshbay-node/src/meshbay_node/ops/roots.py
index c272978..187e134 100644
--- a/packages/meshbay-node/src/meshbay_node/ops/roots.py
+++ b/packages/meshbay-node/src/meshbay_node/ops/roots.py
@@ -1,4 +1,4 @@
-"""A group's directories: adding, removing, changing and ejecting them."""
+"""A group's directories: adding, removing, changing, ejecting and pausing them."""
 
 from __future__ import annotations
 
@@ -126,6 +126,12 @@ async def remove_root(state: dict, group_id: str, root_name: str) -> dict:
     _remove_roots_block(conf_path, group_id, resolved)
 
     cfg.roots.pop(match_idx)
+    # A directory added back later under the same name starts indexed, not
+    # paused by a decision about the one removed here.
+    roster = state.get("roster")
+    if roster:
+        await roster.set_root_paused(group_id, rname, False,
+                                     set_by=state.get("node_user_id", ""))
 
     # Not mutating the live set here either — see `add_root`. Dropping the
     # root from it would leave `retarget` unable to tell that its entries
@@ -304,3 +310,68 @@ async def plug_root(state: dict, group_id: str, root_name: str) -> dict:
     log.info("Root plugged: %s in group %s", root_name, group_id[:8])
     return {"status": "plugged", "name": root_name, "group_id": group_id,
             "roots": roots.describe()}
+
+
+def _live_root(state: dict, group_id: str, root_name: str):
+    """
+    The group's live RootSet and the root named `root_name` in it.
+
+    The indexer's set when the group has no context yet: a group added while
+    the node runs gets one only once its first scan ends, and that scan — a
+    whole new library — is the one most worth pausing.
+    """
+    config = _config(state)
+    if not any(g.id == group_id for g in config.groups):
+        raise OpError("Group not configured on this node", status=404)
+    ctx = state.get("groups_ctx", {}).get(group_id)
+    indexer = state.get("indexers", {}).get(group_id)
+    roots: RootSet | None = (ctx.get("roots") if ctx else None) or (
+        indexer.roots if indexer else None)
+    if not roots:
+        raise OpError("Group has no roots", status=503)
+    root = roots.by_name(root_name)
+    if root is None:
+        raise OpError(f"No root named {root_name!r} in this group", status=404)
+    return roots, root
+
+
+async def pause_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Stop indexing a root; what is already indexed stays listed."""
+    roots, root = _live_root(state, group_id, root_name)
+    if root.paused:
+        return {"status": "already_paused", "name": root.name,
+                "group_id": group_id, "roots": roots.describe()}
+
+    # Persisted first: a restart between the two must not start the scan the
+    # operator stopped.
+    await _roster(state).set_root_paused(
+        group_id, root.name, True, set_by=state.get("node_user_id", ""))
+    # The flag on the context's set too, as `eject_root` does, for a context
+    # whose indexer a retarget replaced.
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        await indexer.pause_root(root.name)
+    root.paused = True
+
+    log.info("Root paused: %s in group %s", root.name, group_id[:8])
+    return {"status": "paused", "name": root.name, "group_id": group_id,
+            "roots": roots.describe()}
+
+
+async def resume_root(state: dict, group_id: str, root_name: str) -> dict:
+    """Index a paused root again; the walk runs in the background."""
+    roots, root = _live_root(state, group_id, root_name)
+    if not root.paused:
+        return {"status": "already_resumed", "name": root.name,
+                "group_id": group_id, "roots": roots.describe()}
+
+    await _roster(state).set_root_paused(
+        group_id, root.name, False, set_by=state.get("node_user_id", ""))
+    indexer = state.get("indexers", {}).get(group_id)
+    if indexer:
+        await indexer.resume_root(root.name)
+    root.paused = False
+
+    log.info("Root resumed: %s in group %s", root.name, group_id[:8])
+    return {"status": "resumed", "name": root.name, "group_id": group_id,
+            "roots": roots.describe()}