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>

3 files changed, 67 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py
index 7b3b1c9..aab3691 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admin.py
@@ -24,8 +24,10 @@ from meshbay_common.adminop import (
     OP_MEMBER_REVOKE,
     OP_MUSICBRAINZ_ENABLED,
     OP_ROOT_EJECT,
+    OP_ROOT_PAUSE,
     OP_ROOT_PLUG,
     OP_ROOT_REMOVE,
+    OP_ROOT_RESUME,
     OP_SEARCH_LISTED,
     OP_SET_SCAN_SETTINGS,
     OP_TMDB_CONFIG,
@@ -72,6 +74,8 @@ _ADMIN_EXECUTORS = {
     OP_CHAT_PURGE: "_admin_exec_chat_purge",
     OP_ROOT_EJECT: "_admin_exec_root_eject",
     OP_ROOT_PLUG: "_admin_exec_root_plug",
+    OP_ROOT_PAUSE: "_admin_exec_root_pause",
+    OP_ROOT_RESUME: "_admin_exec_root_resume",
 }
 
 
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
index 5d4ed94..0ceefb5 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/dispatch.py
@@ -71,6 +71,8 @@ _HANDLERS = {
     MNP.ROOT_REMOVE: ("_do_root_remove", INLINE),
     MNP.ROOT_EJECT: ("_do_root_eject", INLINE),
     MNP.ROOT_PLUG: ("_do_root_plug", INLINE),
+    MNP.ROOT_PAUSE: ("_do_root_pause", INLINE),
+    MNP.ROOT_RESUME: ("_do_root_resume", INLINE),
     MNP.KEYPAIR_BUNDLE_STORE: ("_do_keypair_bundle_store", SPAWNED),
     MNP.KEYPAIR_BUNDLE_DELETE: ("_do_keypair_bundle_delete", SPAWNED),
     MNP.USER_BLOB_STORE: ("_do_user_blob_store", SPAWNED),
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py
index 237a359..c1f424d 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/node_ops.py
@@ -1,5 +1,5 @@
 """The operator's controls over a group's roots and scan pacing that MNP carries:
-removing, ejecting and plugging a root. What widens the sharing, and the node's
+removing, ejecting, plugging, pausing and resuming a root. What widens the sharing, and the node's
 own status, settings, roster and denylist, are the loopback API's and the CLI's
 (MNP 6.0)."""
 
@@ -8,8 +8,10 @@ import logging
 from meshbay_common import MNP_VERSION
 from meshbay_common.adminop import (
     OP_ROOT_EJECT,
+    OP_ROOT_PAUSE,
     OP_ROOT_PLUG,
     OP_ROOT_REMOVE,
+    OP_ROOT_RESUME,
     OP_SET_SCAN_SETTINGS,
 )
 from meshbay_common.protocol import MNP
@@ -201,6 +203,64 @@ class NodeOpsMixin:
             except Exception:
                 pass
 
+    def _do_root_pause(self, msg: dict) -> None:
+        self._ask_root_index_op(msg, OP_ROOT_PAUSE)
+
+    def _do_root_resume(self, msg: dict) -> None:
+        self._ask_root_index_op(msg, OP_ROOT_RESUME)
+
+    def _ask_root_index_op(self, msg: dict, op: str) -> None:
+        target_group = str(msg.get("group_id", self._group_id or "")).strip()
+        root_name = str(msg.get("root_name", "")).strip()
+        if not target_group or not root_name:
+            self._send({"type": "error", "detail": "Missing group_id or root_name"})
+            return
+        if not self._has_admin_authority():
+            self._send({"type": "error", "detail": "No authorized key for this"})
+            return
+        self._issue_admin_challenge(
+            op, root_name,
+            payload={"group_id": target_group, "root_name": root_name},
+            group_id=target_group)
+
+    async def _admin_exec_root_pause(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        await self._exec_root_index_op(
+            pending, transcript, sig, OP_ROOT_PAUSE, ops.pause_root, MNP.ROOT_PAUSE_ACK)
+
+    async def _admin_exec_root_resume(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        await self._exec_root_index_op(
+            pending, transcript, sig, OP_ROOT_RESUME, ops.resume_root,
+            MNP.ROOT_RESUME_ACK)
+
+    async def _exec_root_index_op(self, pending: dict, transcript: bytes, sig: bytes,
+                                  op: str, run, ack: str) -> None:
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"{op}:{pending['subject'][:24]}")
+            return
+        p = pending["payload"]
+        try:
+            result = await self._run_op(run, p["group_id"], p["root_name"])
+        except ops.OpError as e:
+            self._send({"type": "error", "detail": e.message})
+            return
+        except Exception as e:
+            log.error("%s failed: %s", op, e, exc_info=True)
+            self._send({"type": "error", "detail": "Internal error"})
+            return
+        self._audit(op, p["root_name"])
+        # Broadcast, like eject's: every open roots table shows the new state.
+        notice = {"type": ack, "v": MNP_VERSION, **result}
+        for uid, session in list(self._peer_registry().items()):
+            try:
+                session._send(notice)
+            except Exception:
+                pass
+
     async def _retarget_indexer(self, group_id: str) -> None:
         """
         Pick up a root that was just removed from node.toml.