1 files changed, 61 insertions, 1 deletions

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.