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>

4 files changed, 158 insertions, 5 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):
diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py
index 4fffc4e..e0b5518 100644
--- a/packages/meshbay-cli/tests/test_cli_actions.py
+++ b/packages/meshbay-cli/tests/test_cli_actions.py
@@ -425,6 +425,38 @@ def test_browser_access_is_set_here_and_mirrored_never_taken_from_the_hub(
     assert store.state["browser_access"] == "off"
 
 
+def test_rm_deletes_only_what_this_account_may_and_asks_first(node, monkeypatch, capsys):
+    from fake_hub import USER_ID
+    mine = node.add("share", "mine.txt", b"1", uploader_id=USER_ID)
+    node.add("share", "also-mine.txt", b"2", uploader_id=USER_ID)
+    theirs = node.add("share", "theirs.txt", b"3", uploader_id="someone-else")
+    node.dirs.add("share/sub")
+
+    # Somebody else's file, a folder: refused whole, before any question.
+    assert main("rm", "--yes", "Family/share/mine.txt", "Family/share/theirs.txt") == 8
+    assert "Family@alice-test/share/theirs.txt" in capsys.readouterr().err
+    assert main("rm", "--yes", "Family/share/sub") == 1
+    assert main("rm", "--yes", "Family/share/nothing*") == 3
+    assert mine["id"] in node.files
+
+    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
+    assert main("rm", "Family/share/mine.txt") == 1                  # not confirmed
+    monkeypatch.setattr("sys.stdin", _Terminal("n\n"))
+    assert main("rm", "Family/share/mine.txt") == 1
+    assert mine["id"] in node.files
+    monkeypatch.setattr("sys.stdin", _Terminal("y\n"))
+    assert main("rm", "Family/share/*mine.txt") == 0
+    assert "Deleted 2 files" in capsys.readouterr().err
+    assert set(node.files) == {theirs["id"]}
+
+    # From the node's operator key, anyone's file.
+    node.role = "operator"
+    assert main("--json", "rm", "--yes", "Family/share/theirs.txt") == 0
+    assert json.loads(capsys.readouterr().out)["deleted"] == [
+        "Family@alice-test/share/theirs.txt"]
+    assert not node.files
+
+
 def test_every_action_is_run_here():
     """Runs last in this module: every action the parser knows was run above."""
     assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}"
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
index 9cac392..988fb38 100644
--- a/packages/meshbay-cli/tests/test_e2e_local.py
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -529,12 +529,14 @@ def test_a_machine_with_no_identity_there_is_linked_as_a_device(world):
 
 
 async def test_the_interactive_mode_against_the_real_node(world, monkeypatch, tmp_path):
-    """The panes, opened on a real group, and F5 bringing a file back."""
+    """The panes, opened on a real group: F5 bringing a file back, F5 sending one,
+    and Delete taking that one away again."""
     from meshbay_cli import __main__ as main_module
     from meshbay_cli.commands.context import Context
     from meshbay_cli.tui.app import MeshBayApp
-    from meshbay_cli.tui.panes import RemotePane
+    from meshbay_cli.tui.panes import LocalPane, RemotePane
     from meshbay_cli.tui.screens import MainScreen
+    from meshbay_cli.tui.widgets import Choice
 
     # Browser access off, whichever test ran before: this machine must be linked.
     await world["app"].withdraw_bundle()
@@ -585,6 +587,26 @@ async def test_the_interactive_mode_against_the_real_node(world, monkeypatch, tm
         await until(pilot, lambda: (tmp_path / "notes.txt").exists()
                     and (tmp_path / "notes.txt").read_text() == "hello\n")
 
+        # Sent from the left, then deleted from the right: it was this
+        # account's upload, so the node lets it go, from its disk too.
+        (tmp_path / "tui-upload.txt").write_text("sent from the panes\n")
+        local = app.screen.query_one(LocalPane)
+        local.action_refresh()
+        await until(pilot, lambda: "tui-upload.txt" in names(local))
+        local.table.move_cursor(row=names(local).index("tui-upload.txt"))
+        local.table.focus()
+        await pilot.press("f5")
+        await until(pilot, lambda: "tui-upload.txt" in names(remote))
+        assert (world["share"] / "tui-upload.txt").exists()
+        remote.table.move_cursor(row=names(remote).index("tui-upload.txt"))
+        remote.table.focus()
+        await pilot.press("delete")
+        await until(pilot, lambda: isinstance(app.screen, Choice))
+        await pilot.click("#choice-delete")
+        await until(pilot, lambda: "tui-upload.txt" not in names(remote))
+        assert not (world["share"] / "tui-upload.txt").exists()
+        assert (world["share"] / "notes.txt").exists()
+
 
 # ── The application's half of linking, in a real Chrome ──────────────────────
 #
@@ -833,6 +855,24 @@ def test_a_member_joins_links_devices_shares_with_browsers_and_leaves(world):
     assert r.returncode == 0, r.stderr
     assert bob(world, "bob3", "ls", group).returncode == 5      # a stranger again
 
+    # rm: what bob uploaded, from another of his machines too; nothing of the
+    # owner's, whose share bob did not upload.
+    up = world["root"] / "bob-upload.txt"
+    up.write_text("from bob\n")
+    assert bob(world, "bob1", "put", str(up), f"{group}/share").returncode == 0
+    deadline = time.monotonic() + 30          # indexed once the node has written it
+    while "bob-upload.txt" not in bob(world, "bob1", "ls", f"{group}/share").stdout:
+        assert time.monotonic() < deadline, "the upload never reached the index"
+        time.sleep(1)
+    r = bob(world, "bob1", "rm", f"{group}/share/bob-upload.txt")
+    assert r.returncode == 1 and "--yes" in r.stderr             # no terminal to ask on
+    r = bob(world, "bob1", "rm", "--yes", f"{group}/share/notes.txt")
+    assert r.returncode == 8 and "did not upload" in r.stderr, r.stderr
+    r = bob(world, "bob2", "rm", "--yes", f"{group}/share/bob-upload.txt")
+    assert r.returncode == 0, r.stderr
+    assert "bob-upload.txt" not in bob(world, "bob1", "ls", f"{group}/share").stdout
+    assert (world["share"] / "notes.txt").exists()
+
     # Leaving: asked first, then the group is gone from the account.
     assert bob(world, "bob1", "leave", group).returncode == 1
     r = bob(world, "bob1", "leave", "--yes", group)
diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py
index 680b9e0..49180d5 100644
--- a/packages/meshbay-cli/tests/test_interactive.py
+++ b/packages/meshbay-cli/tests/test_interactive.py
@@ -157,3 +157,52 @@ async def test_a_read_only_folder_takes_no_upload(world):
         await pilot.pause(0.3)
         assert node.stored("share/mine.txt") is None
         assert any("does not accept uploads" in n.message for n in app._notifications)
+
+
+async def test_delete_removes_what_is_yours_and_asks_first(world):
+    from fake_hub import USER_ID
+    hub, node, work = world
+    mine = node.add("share", "mine.txt", b"1", uploader_id=USER_ID)
+    theirs = node.add("share", "theirs.txt", b"2", uploader_id="someone-else")
+    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
+                                              "--password", PASSPHRASE, "login"]) == 0
+    app = _app()
+    async with app.run_test(size=(120, 30)) as pilot:
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        remote = app.screen.query_one(RemotePane)
+        await _open(pilot, remote, f"Family@{USERNAME}/")
+        await _open(pilot, remote, "share/")
+        await _until(pilot, lambda: "theirs.txt" in _names(remote))
+
+        # Somebody else's file, marked with one's own: nothing is deleted.
+        for name in ("mine.txt", "theirs.txt"):
+            remote.table.move_cursor(row=_names(remote).index(name))
+            await pilot.press("space")
+        await pilot.press("delete")
+        await _until(pilot, lambda: isinstance(app.screen, Choice))
+        assert "theirs.txt" in str(app.screen.message)
+        await pilot.click("#choice-ok")
+        assert mine["id"] in node.files and theirs["id"] in node.files
+
+        # One's own: asked, cancelled, asked again, deleted, and gone from the pane.
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        remote.marks.clear()
+        remote.redraw()
+        remote.table.move_cursor(row=_names(remote).index("mine.txt"))
+        await pilot.press("delete")
+        await _until(pilot, lambda: isinstance(app.screen, Choice))
+        await pilot.click("#choice-cancel")
+        assert mine["id"] in node.files
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        await pilot.press("delete")
+        await _until(pilot, lambda: isinstance(app.screen, Choice))
+        await pilot.click("#choice-delete")
+        await _until(pilot, lambda: mine["id"] not in node.files)
+        await _until(pilot, lambda: "mine.txt" not in _names(remote))
+        assert theirs["id"] in node.files
+
+        # On the left, Delete removes nothing of this machine's.
+        app.screen.query_one(LocalPane).table.focus()
+        await pilot.press("delete")
+        await pilot.pause(0.2)
+        assert any("select them on the right" in n.message for n in app._notifications)