feat(cli): rm, and Delete in the interactive mode

rm deletes files, resolved as get resolves paths and patterns, with the
node's rule (design §5.4): the uploading account's devices, or the node's
operator account. The whole run is checked against the index first (the
entry's uploader_id, the role the join result gives this key): a file
outside the rule, or a folder, refuses it before anything is asked or sent.
Then one confirmation, or --yes. The node's challenge is checked against
what was asked before it is signed. In the interactive mode, Delete on the
right pane does the same, asked once per batch.

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

1 files changed, 35 insertions, 3 deletions

diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index f53eb01..46ad68e 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -22,6 +22,7 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from fake_hub import USER_ID
 from meshbay_cli.client.keys import Identity
 from meshbay_cli.client.session import GroupSession, NodeRefusal
+from meshbay_common.adminop import OP_FILE_DELETE, admin_transcript
 from meshbay_common.device import (
     device_add_transcript,
     device_code_hash,
@@ -57,11 +58,14 @@ class FakeNode:
                          "label": "", "pinned_at": "2026-10-01T10:00:00+00:00",
                          "pinned_via": "code"}]
         self.requests: list[dict] = []
+        self.role = ""                       # "operator": this key was paired here
+        self.admin_ops: dict[str, dict] = {}
 
-    def add(self, path: str, name: str, data: bytes) -> dict:
+    def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
         fid = hashlib.blake2b(data + path.encode() + name.encode(), digest_size=32).hexdigest()
         entry = {"id": fid, "name": name, "path": path, "size": len(data),
-                 "type": "document", "added_at": 1_700_000_000, "data": data}
+                 "type": "document", "added_at": 1_700_000_000, "data": data,
+                 "uploader_id": uploader_id}
         self.files[fid] = entry
         parts = path.split("/")
         for i in range(1, len(parts) + 1):
@@ -70,7 +74,7 @@ class FakeNode:
 
     def session(self) -> GroupSession:
         return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek,
-                            {}, self.identity, self.nonce)
+                            {}, self.identity, self.nonce, self.role)
 
     def request_device(self, code: str) -> Identity:
         """Another machine files a request with `code`, as it would before its proof."""
@@ -194,6 +198,34 @@ class FakeChannel:
                                     dir=folder)
 
 
+    # ── Deleting: a challenge, signed with this machine's key ──────────────
+
+    def _file_delete(self, msg):
+        n = self.node
+        if msg["file_id"] not in n.files:
+            return {"type": "error", "detail": "File not found"}
+        op_id, nonce = secrets.token_hex(8), secrets.token_bytes(32)
+        n.admin_ops[op_id] = {"subject": msg["file_id"], "nonce": nonce, "ts": 1_700_000_000}
+        return {"type": "admin_challenge", "op_id": op_id, "op": OP_FILE_DELETE,
+                "subject": msg["file_id"], "nonce": base64.b64encode(nonce).decode(),
+                "ts": 1_700_000_000, "node_pk": NODE_PK, "group_id": GROUP_ID}
+
+    def _admin_response(self, msg):
+        n = self.node
+        op = n.admin_ops.pop(msg["op_id"], None)
+        if op is None:
+            return {"type": "error", "detail": "No pending admin operation"}
+        entry = n.files.get(op["subject"])
+        transcript = admin_transcript(op=OP_FILE_DELETE, node_pk_b64=NODE_PK,
+                                      group_id=GROUP_ID, subject=op["subject"],
+                                      nonce=op["nonce"], ts=op["ts"])
+        signed = _signed_by_a_device(n, transcript, msg["signature"])
+        allowed = n.role == "operator" or (entry and entry["uploader_id"] == USER_ID)
+        if not (signed and allowed):
+            return {"type": "error", "detail": "Signature verification failed"}
+        del n.files[op["subject"]]
+        return {"type": "file_delete_ack", "file_id": op["subject"]}
+
     # ── Devices ──────────────────────────────────────────────────────────────
 
     def _device_list(self, msg):