feat(cli): meshbay-cli, a command-line client for members

Signs in to the hub, lists the account's groups and their folders, and
downloads and uploads files directly with each group's node, end-to-end
encrypted. With no argument it opens a two-pane interactive mode (Textual).
Session and device keys are kept encrypted under the system's key storage,
or a 0600 key file where there is none; node keys are pinned at first
contact; a new machine links as a device.

Packaged on its own on Linux (deb and rpm, on meshbay-common's venv, with a
man page generated from the parser) and frozen next to meshbay-node in the
full Windows installer, not the light one, with a self-test the build runs.
client/keys.py ports keyderive.js and is LGPL like it.

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

12 files changed, 2008 insertions, 0 deletions

diff --git a/packages/meshbay-cli/tests/conftest.py b/packages/meshbay-cli/tests/conftest.py
new file mode 100644
index 0000000..8ef9189
--- /dev/null
+++ b/packages/meshbay-cli/tests/conftest.py
@@ -0,0 +1,49 @@
+"""
+Every test runs with its own home, and never touches the developer's own.
+
+Redirecting HOME alone isolates nothing on Windows — `Path.home()` reads
+USERPROFILE there, and the CLI's directories come from LOCALAPPDATA — so all of
+them are pointed at a throwaway directory. The OS key storage is replaced by
+keyring's refusing backend: a test must never write into the developer's real
+credential store, and the store's file fallback is what a headless machine uses
+anyway. A test that needs a working key storage installs its own backend.
+"""
+
+import logging
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def isolated_home(tmp_path, monkeypatch):
+    home = tmp_path / "home"
+    home.mkdir()
+    for var in ("HOME", "USERPROFILE"):
+        monkeypatch.setenv(var, str(home))
+    for var, sub in (("APPDATA", "AppData/Roaming"), ("LOCALAPPDATA", "AppData/Local"),
+                     ("XDG_CONFIG_HOME", ".config"), ("XDG_DATA_HOME", ".local/share")):
+        monkeypatch.setenv(var, str(home / sub))
+    for var in ("MESHBAY_CLI_HOME", "MESHBAY_HUB", "MESHBAY_USER", "MESHBAY_PASSWORD"):
+        monkeypatch.delenv(var, raising=False)
+    monkeypatch.setenv("PYTHON_KEYRING_BACKEND", "keyring.backends.fail.Keyring")
+    return home
+
+
+@pytest.fixture(autouse=True)
+def restored_logging():
+    """`main()` configures logging, as an entry point does; a test that calls it
+    in-process must not leave that configuration to the next package's tests."""
+    root = logging.getLogger()
+    names = ("meshbay_cli", "aiortc", "aioice", "httpx", "httpcore")
+    saved = (list(root.handlers), root.level, root.disabled,
+             {n: logging.getLogger(n).level for n in names})
+    yield
+    handlers, level, disabled, levels = saved
+    for h in root.handlers:
+        if h not in handlers:
+            h.close()
+    root.handlers[:] = handlers
+    root.setLevel(level)
+    root.disabled = disabled
+    for n, lv in levels.items():
+        logging.getLogger(n).setLevel(lv)
diff --git a/packages/meshbay-cli/tests/fake_hub.py b/packages/meshbay-cli/tests/fake_hub.py
new file mode 100644
index 0000000..ec1a0c0
--- /dev/null
+++ b/packages/meshbay-cli/tests/fake_hub.py
@@ -0,0 +1,108 @@
+"""
+A hub that keeps the rules the CLI depends on, in memory.
+
+Its refresh tokens **rotate, and a spent one presented again revokes the
+whole family**, as the real hub does. A lax stub would pass a client that
+renews twice with the same token — which is exactly the failure two runs at
+once produce.
+"""
+
+import asyncio
+import base64
+import secrets
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+from meshbay_cli.client.hub import HubError
+from meshbay_cli.client.keys import auth_key
+
+USERNAME = "alice-test"
+PASSPHRASE = "a passphrase for the fake hub"
+USER_ID = "0b4f6f0e-5d7e-4e8a-9d2b-6a1c1b9e2f11"
+PEPPER = base64.b64encode(b"\x07" * 32).decode()
+
+
+class FakeHub:
+    def __init__(self, url: str = "https://hub.test"):
+        self.url = url
+        self.ak = auth_key(USERNAME, PASSPHRASE)
+        self.refresh_tokens: dict[str, dict] = {}   # token → {family, spent}
+        self.revoked_families: set[str] = set()
+        self.access: set[str] = set()
+        self.devices: dict[str, str] = {}           # id → pk
+        self.refresh_calls = 0
+        self.reuse_detected = False
+        self.refresh_delay = 0.0
+        self.groups = [{"id": "g1", "name": "Family", "owner_username": USERNAME,
+                        "node_online": True, "visibility": "private", "is_admin": True}]
+
+    def _session(self, family: str | None = None) -> dict:
+        family = family or secrets.token_hex(4)
+        rt, at = secrets.token_hex(8), secrets.token_hex(8)
+        self.refresh_tokens[rt] = {"family": family, "spent": False}
+        self.access.add(at)
+        return {"access_token": at, "refresh_token": rt, "expires_in": 14400}
+
+    def _auth(self, token: str) -> None:
+        if token not in self.access:
+            raise HubError(401, "Invalid token")
+
+    async def aclose(self) -> None:
+        pass
+
+    async def check_version(self) -> None:
+        return None
+
+    async def login(self, username, ak, known_browser):
+        from meshbay_cli.client.hub import _sign_in_refusal
+        if username != USERNAME or ak != self.ak:
+            raise _sign_in_refusal(HubError(401, "Invalid credentials"))
+        return {**self._session(), "bundle_pepper": PEPPER, "bundle_pepper_version": 1,
+                "known_browser": "kb-1"}
+
+    async def me(self, token):
+        self._auth(token)
+        return {"user_id": USER_ID, "username": USERNAME}
+
+    async def register_device(self, token, pk, label, ak):
+        self._auth(token)
+        assert ak == self.ak, "registering a device takes the passphrase"
+        for did, known in self.devices.items():
+            if known == pk:
+                return {"id": did, "existing": True}
+        did = f"dev-{len(self.devices) + 1}"
+        self.devices[did] = pk
+        return {"id": did, "existing": False}
+
+    async def delete_device(self, token, device_id):
+        self._auth(token)
+        self.devices.pop(device_id)
+
+    async def device_auth(self, username, sk):
+        pk = base64.b64encode(sk.public_key().public_bytes_raw()).decode()
+        if pk not in self.devices.values():
+            from meshbay_cli.errors import AUTH, CliError
+            raise CliError(AUTH, "Sign-in refused")
+        Ed25519PublicKey.from_public_bytes(sk.public_key().public_bytes_raw())
+        return {**self._session(), "bundle_pepper": PEPPER}
+
+    async def refresh(self, rt):
+        self.refresh_calls += 1
+        row = self.refresh_tokens.get(rt)
+        if row is None or row["family"] in self.revoked_families:
+            raise HubError(401, "Invalid refresh token")
+        if row["spent"]:
+            self.reuse_detected = True
+            self.revoked_families.add(row["family"])
+            raise HubError(401, "Token reuse detected — family revoked")
+        row["spent"] = True
+        await asyncio.sleep(self.refresh_delay)
+        return self._session(row["family"])
+
+    async def logout(self, rt):
+        row = self.refresh_tokens.get(rt)
+        if row:
+            self.revoked_families.add(row["family"])
+
+    async def my_groups(self, token):
+        self._auth(token)
+        return self.groups
diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
new file mode 100644
index 0000000..e733c56
--- /dev/null
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -0,0 +1,166 @@
+"""
+A node in memory, at the level of the MNP messages the CLI sends.
+
+It does the real cryptography — chunks encrypted under the per-chunk key,
+uploads unsealed under the group key, acks sealed back — and keeps the rules
+that shape the client: a lease before any chunk, chunks of an upload in order,
+the position an interrupted upload resumes from, no overwrite. Not a model of
+the node's code: a node the CLI must work against.
+"""
+
+import asyncio
+import hashlib
+import math
+import secrets
+
+from meshbay_cli.client.session import GroupSession, NodeRefusal
+from meshbay_common.groupbox import PURPOSE_INDEX, PURPOSE_UPLOAD, seal, unseal
+from meshbay_common.protocol import UPLOAD_PROBE_INDEX, file_upload_ack_wire
+from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
+
+GROUP_ID = "g1"
+CHUNK = 1024 * 1024
+
+
+class FakeNode:
+    def __init__(self, *, writable: bool = True):
+        self.gek = secrets.token_bytes(32)
+        self.roots = [{"name": "share", "available": True, "writable": writable}]
+        self.dirs = {"share"}
+        self.files: dict[str, dict] = {}     # id → entry, with "data"
+        self.leases: dict[str, str] = {}     # tr → state
+        self.queue_first = 0                 # leases to answer "queued" before granting
+        self.uploads: dict[tuple, dict] = {} # (dir, name) → {"next", "data"}
+        self.fail_chunk: tuple[str, int] | None = None   # (file_id, index) → refused
+        self.refuse_upload_code = ""
+        self.chunk_requests = 0
+        self.chunk_delay = 0.0
+
+    def add(self, path: str, name: str, data: bytes) -> 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}
+        self.files[fid] = entry
+        parts = path.split("/")
+        for i in range(1, len(parts) + 1):
+            self.dirs.add("/".join(parts[:i]))
+        return entry
+
+    def session(self) -> GroupSession:
+        return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", "pk", self.gek, {})
+
+    def stored(self, path: str) -> bytes | None:
+        parent, _, name = path.rpartition("/")
+        for e in self.files.values():
+            if e["path"] == parent and e["name"] == name:
+                return e["data"]
+        return None
+
+
+class FakeChannel:
+    def __init__(self, node: FakeNode):
+        self.node = node
+        self.closed = asyncio.Event()
+        self._push: dict[str, list] = {}
+
+    def on_push(self, t, h):
+        self._push.setdefault(t, []).append(h)
+
+    def off_push(self, t, h):
+        if h in self._push.get(t, []):
+            self._push[t].remove(h)
+
+    def _emit(self, msg):
+        for h in list(self._push.get(msg["type"], [])):
+            h(msg)
+
+    async def request(self, msg: dict, *expect: str, timeout: float = 60) -> dict:
+        await asyncio.sleep(self.node.chunk_delay if msg["type"] == "file_req" else 0)
+        reply = getattr(self, "_" + msg["type"])(msg)
+        if reply.get("type") == "error":
+            raise NodeRefusal(reply["detail"], reply.get("code", ""))
+        return reply
+
+    # ── Messages ─────────────────────────────────────────────────────────────
+
+    def _index_sync(self, msg):
+        n = self.node
+        payload = {"version": 1, "roots": n.roots, "dirs": sorted(n.dirs),
+                   "entries": [{k: v for k, v in e.items() if k != "data"}
+                               for e in n.files.values()]}
+        return {"type": "index_sync", "group_id": GROUP_ID,
+                **seal(n.gek, PURPOSE_INDEX, "index_sync", GROUP_ID, payload)}
+
+    def _transfer_open(self, msg):
+        tr = msg["tr"]
+        if self.node.queue_first:
+            self.node.queue_first -= 1
+            self.node.leases[tr] = "queued"
+            asyncio.get_running_loop().call_later(0.01, self._grant, tr)
+            return {"type": "transfer_state", "tr": tr, "state": "queued", "ahead": 1}
+        self.node.leases[tr] = "granted"
+        return {"type": "transfer_state", "tr": tr, "state": "granted"}
+
+    def _grant(self, tr):
+        self.node.leases[tr] = "granted"
+        self._emit({"type": "transfer_state", "tr": tr, "state": "granted"})
+
+    def _transfer_close(self, msg):
+        self.node.leases.pop(msg["tr"], None)
+        return {"type": "transfer_state", "tr": msg["tr"], "state": "closed"}
+
+    def _file_req(self, msg):
+        n = self.node
+        n.chunk_requests += 1
+        if n.leases.get(msg.get("tr")) != "granted":
+            return {"type": "error", "detail": "no lease", "code": "lease_not_granted"}
+        if n.fail_chunk == (msg["file_id"], msg["chunk_index"]):
+            return {"type": "error", "detail": "The drive stopped answering"}
+        entry = n.files[msg["file_id"]]
+        i = msg["chunk_index"]
+        data = entry["data"][i * CHUNK:(i + 1) * CHUNK]
+        nonce, ct = encrypt_chunk_aes(chunk_key_aes(n.gek, bytes.fromhex(entry["id"]), i), data)
+        return {"type": "file_chunk", "chunk_index": i, "nonce": nonce, "ct": ct}
+
+    def _dir_create(self, msg):
+        root = self.node.roots[0]
+        if not root["writable"]:
+            return {"type": "error", "detail": "read-only", "code": "root_read_only"}
+        path = f"{msg['dir']}/{msg['name']}"
+        self.node.dirs.add(path)
+        return {"type": "dir_create_ack", "dir": path}
+
+    def _file_upload(self, msg):
+        n = self.node
+        if n.leases.get(msg.get("tr")) != "granted":
+            return {"type": "error", "detail": "no lease", "code": "lease_not_granted"}
+        p = unseal(n.gek, PURPOSE_UPLOAD, "file_upload", GROUP_ID, msg)
+        name, folder, i = p["filename"], p["dir"], msg["chunk_index"]
+        if n.refuse_upload_code:
+            return {"type": "error", "detail": "refused", "code": n.refuse_upload_code}
+        key = (folder, name)
+        state = n.uploads.get(key)
+        stored = name
+        while n.stored(f"{folder}/{stored}") is not None:
+            stem, dot, ext = name.partition(".")
+            stored = f"{stem} (1){dot}{ext}"
+        if i == UPLOAD_PROBE_INDEX:
+            return file_upload_ack_wire(n.gek, GROUP_ID, upload_id=msg["upload_id"],
+                                        chunk_index=i, filename=name, stored_as="",
+                                        dir=folder, resume_from=state["next"] if state else 0)
+        if state is None:
+            state = n.uploads[key] = {"next": 0, "data": b""}
+        if i != state["next"]:
+            return {"type": "error", "detail": "Out of order", "code": "bad_chunk_index"}
+        state["data"] += p["data"]
+        state["next"] += 1
+        if state["next"] >= msg["total_chunks"]:
+            del n.uploads[key]
+            n.add(folder, stored, state["data"])
+        return file_upload_ack_wire(n.gek, GROUP_ID, upload_id=msg["upload_id"],
+                                    chunk_index=i, filename=name, stored_as=stored,
+                                    dir=folder)
+
+
+def chunks_of(size: int, chunk: int) -> int:
+    return max(1, math.ceil(size / chunk))
diff --git a/packages/meshbay-cli/tests/test_cli_actions.py b/packages/meshbay-cli/tests/test_cli_actions.py
new file mode 100644
index 0000000..b88d93c
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_cli_actions.py
@@ -0,0 +1,238 @@
+"""
+Every action, run through `main()` with the hub and the node stubbed.
+
+A branch nobody has run is covered by nothing — the node's `reload` once shipped
+with a missing import. `test_every_action_is_run_here` refuses an action added
+to the parser without a run below.
+"""
+
+import io
+import json
+
+import pytest
+from fake_hub import PASSPHRASE, USERNAME, FakeHub
+from meshbay_cli import __main__ as cli
+from meshbay_cli.client import pool
+from meshbay_cli.commands import context
+
+RUN: set[str] = set()
+
+
+@pytest.fixture
+def hub(monkeypatch):
+    fake = FakeHub()
+    monkeypatch.setattr(context, "Hub", lambda url: fake)
+    return fake
+
+
+def main(*argv: str) -> int:
+    RUN.add(next(a for a in argv if a in cli.ACTIONS))
+    return cli.main(list(argv))
+
+
+class FakeSession:
+    async def index(self):
+        return {"roots": [{"name": "share", "available": True, "writable": True}],
+                "dirs": ["share"],
+                "entries": [{"id": "a" * 64, "name": "notes.txt", "path": "share",
+                             "size": 6, "type": "document", "added_at": 0}]}
+
+    async def close(self):
+        pass
+
+
+def test_login_ls_logout(hub, monkeypatch, capsys):
+    monkeypatch.setattr(pool, "open_group", lambda account, group: _session())
+
+    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
+    assert "Signed in: alice-test" in capsys.readouterr().out
+
+    assert main("ls") == 0
+    assert capsys.readouterr().out.strip() == f"Family@{USERNAME}"
+
+    assert main("ls", "--json", "Family/share") == 0
+    [item] = json.loads(capsys.readouterr().out)
+    assert item["path"] == f"Family@{USERNAME}/share/notes.txt" and item["kind"] == "file"
+
+    assert main("logout") == 0
+    assert "Signed out." in capsys.readouterr().out
+    assert not hub.devices
+
+    assert main("ls") == 2
+    assert "Not signed in" in capsys.readouterr().err
+
+
+def test_login_without_remembering_stores_nothing(hub, isolated_home, capsys):
+    assert main("--hub", hub.url, "-u", USERNAME, "login", "--no-remember",
+                "--password", PASSPHRASE) == 0
+    assert "Credentials accepted" in capsys.readouterr().out
+    assert not hub.devices
+    assert not any(isolated_home.rglob("secrets.enc"))
+
+
+def test_a_wrong_passphrase_exits_2(hub, capsys):
+    assert main("--hub", hub.url, "-u", USERNAME, "--password", "nope", "login") == 2
+    assert "wrong username or passphrase" in capsys.readouterr().err
+
+
+def test_no_passphrase_and_no_terminal_exits_2(hub, monkeypatch, capsys):
+    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
+    assert main("--hub", hub.url, "-u", USERNAME, "login") == 2
+    assert "MESHBAY_PASSWORD" in capsys.readouterr().err
+
+
+async def _session():
+    return FakeSession()
+
+
+@pytest.fixture
+def node(hub, monkeypatch, capsys):
+    from fake_node import FakeNode
+    fake = FakeNode()
+
+    async def open_group(account, group):
+        return fake.session()
+    monkeypatch.setattr(pool, "open_group", open_group)
+    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
+    capsys.readouterr()
+    return fake
+
+
+def test_get_files_folders_and_patterns(node, tmp_path, monkeypatch, capsys):
+    node.add("share", "notes.txt", b"hello")
+    node.add("share/2024", "a.jpg", b"A" * 3000)
+    node.add("share/2024", "b.jpg", b"B" * 10)
+    node.add("share/2024/deep", "c.txt", b"C")
+    node.dirs.add("share/2024/empty")
+    monkeypatch.chdir(tmp_path)
+
+    assert main("get", "Family/share/notes.txt") == 0
+    assert (tmp_path / "notes.txt").read_bytes() == b"hello"
+    assert "Downloaded 1 file" in capsys.readouterr().err
+
+    # Already there: refused by default, the run fails at the end; -f overwrites.
+    (tmp_path / "notes.txt").write_bytes(b"mine")
+    assert main("get", "Family/share/notes.txt") == 4
+    assert "already exists" in capsys.readouterr().err
+    assert (tmp_path / "notes.txt").read_bytes() == b"mine"
+    assert main("get", "-f", "Family/share/notes.txt") == 0
+    assert (tmp_path / "notes.txt").read_bytes() == b"hello"
+
+    assert main("get", "Family/share/2024/*.jpg", "-o", "pics/") == 0
+    assert sorted(p.name for p in (tmp_path / "pics").iterdir()) == ["a.jpg", "b.jpg"]
+
+    assert main("get", "Family/share/2024") == 1          # a folder needs -R
+    assert "use -R" in capsys.readouterr().err
+    assert main("get", "-R", "Family/share/2024", "-o", "tree/") == 0
+    assert (tmp_path / "tree/2024/deep/c.txt").read_bytes() == b"C"
+    assert (tmp_path / "tree/2024/empty").is_dir()
+
+    assert main("get", "Family/share/nope*") == 3
+
+    assert main("get", "Family/share/notes.txt", "-o", "-") == 0
+    assert capsys.readouterr().out == "hello"
+
+
+def test_put_files_folders_and_stdin(node, tmp_path, monkeypatch, capsys):
+    node.add("share", "taken.txt", b"theirs")
+    (tmp_path / "a.txt").write_bytes(b"one")
+    (tmp_path / "Taken.TXT").write_bytes(b"mine")
+    scans = tmp_path / "Scans"
+    (scans / "2024").mkdir(parents=True)
+    (scans / "2024" / "p1.pdf").write_bytes(b"%PDF" + b"x" * 100_000)
+    (scans / "empty").mkdir()
+
+    assert main("put", str(tmp_path / "a.txt"), "Family/share") == 0
+    assert node.stored("share/a.txt") == b"one"
+
+    # A name already there, whatever its case: nothing is sent at all.
+    assert main("put", str(tmp_path / "a.txt"), str(tmp_path / "Taken.TXT"),
+                "Family/share") == 8
+    assert "already there" in capsys.readouterr().err
+    assert node.stored("share/Taken.TXT") is None
+    # -n skips it and sends the rest.
+    assert main("put", "-n", str(tmp_path / "Taken.TXT"), "Family/share") == 0
+
+    assert main("put", str(scans), "Family/share") == 1      # a folder needs -R
+    assert main("put", "-R", str(scans), "Family/share") == 0
+    assert node.stored("share/Scans/2024/p1.pdf").startswith(b"%PDF")
+    assert "share/Scans/empty" in node.dirs
+
+    bad = tmp_path / ".hidden"
+    bad.write_bytes(b"x")
+    assert main("put", str(bad), "Family/share") == 8
+    assert node.stored("share/.hidden") is None
+
+    monkeypatch.setattr("sys.stdin", io.TextIOWrapper(io.BytesIO(b"piped bytes")))
+    assert main("put", "-", "Family/share/from-stdin.txt") == 0
+    assert node.stored("share/from-stdin.txt") == b"piped bytes"
+
+
+def test_put_into_a_read_only_folder_sends_nothing(node, tmp_path, capsys):
+    node.roots[0]["writable"] = False
+    (tmp_path / "a.txt").write_bytes(b"one")
+    assert main("put", str(tmp_path / "a.txt"), "Family/share") == 8
+    assert "read-only" in capsys.readouterr().err
+
+
+def test_trust_lists_and_forgets_pins(hub, capsys):
+    assert main("trust", "list") == 0                    # nothing stored: nothing pinned
+    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
+                "--json", "login") == 0
+    assert json.loads(capsys.readouterr().out)["groups"] == 1
+    from meshbay_cli.client.store import Store
+    store = Store()
+    store.load()
+    store.state["pins"] = {"node-1": "pk-1"}
+    store.save()
+    assert main("--json", "trust", "list") == 0
+    assert json.loads(capsys.readouterr().out) == {"node-1": "pk-1"}
+    assert main("trust", "reset") == 1
+    assert main("trust", "reset", "node-2") == 3
+    assert main("trust", "reset", "node-1") == 0
+    assert main("--json", "trust", "list") == 0
+    capsys.readouterr()
+    assert main("--json", "logout") == 0
+    assert json.loads(capsys.readouterr().out) == {"signed_out": True}
+
+
+def test_a_usage_error_is_exit_1_not_the_sign_in_code(capsys):
+    """argparse's own 2 would read as a refused sign-in to a script."""
+    assert cli.main(["ls", "--no-such-option"]) == 1
+    assert cli.main(["no-such-action"]) == 1
+
+
+
+def test_groups_lists_each_group_with_its_state(hub, capsys):
+    hub.groups.append({"id": "g2", "name": "Club/Photos", "owner_username": "bob-test",
+                       "node_online": False, "visibility": "public", "is_admin": False,
+                       "hosted": True, "last_activity_at": "2026-10-01T12:00:00+00:00"})
+    hub.groups.append({"id": "g3", "name": "Draft\x1b[31m", "owner_username": USERNAME,
+                       "visibility": "private", "is_admin": True, "hosted": False})
+    assert main("groups") == 2                           # signed in nowhere
+    capsys.readouterr()
+    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
+    capsys.readouterr()
+
+    assert main("groups") == 0
+    header, *lines = capsys.readouterr().out.splitlines()
+    assert header.split() == ["NODE", "ROLE", "VISIBILITY", "LAST", "ACTIVITY", "GROUP"]
+    # In name order; a "/" in a name is encoded so the path can be typed back.
+    assert lines[0].split()[:3] == ["down", "member", "public"]
+    assert lines[0].endswith("Club%2FPhotos@bob-test")
+    assert "2026-10-0" in lines[0]
+    assert lines[1].split()[:3] == ["none", "owner", "private"]
+    assert "\x1b" not in lines[1] and lines[1].endswith(f"Draft\\x1b[31m@{USERNAME}")
+    assert lines[2].split()[:3] == ["up", "owner", "private"]
+
+    assert main("--json", "groups") == 0
+    rows = json.loads(capsys.readouterr().out)
+    assert [(r["group"], r["node"], r["role"]) for r in rows] == [
+        ("Club/Photos", "down", "member"), ("Draft\x1b[31m", "none", "owner"),
+        ("Family", "up", "owner")]
+    assert rows[0]["path"] == "Club%2FPhotos@bob-test"
+
+
+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_cli_transfers.py b/packages/meshbay-cli/tests/test_cli_transfers.py
new file mode 100644
index 0000000..c240896
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_cli_transfers.py
@@ -0,0 +1,206 @@
+"""Downloads and uploads against a node in memory (`fake_node.py`)."""
+
+import io
+import json
+import secrets
+
+import pytest
+from fake_node import CHUNK, FakeNode
+from meshbay_cli.client.names import UnsafeName, local_path, upload_name_problem
+from meshbay_cli.client.transfers import (
+    PART_SUFFIX,
+    UPLOAD_CHUNK,
+    TransferError,
+    create_dir,
+    download,
+    upload,
+)
+
+
+def _reader(data: bytes):
+    buf = io.BytesIO(data)
+
+    def read(offset, length):
+        buf.seek(offset)
+        return buf.read(length)
+    return read
+
+
+# ── Download ─────────────────────────────────────────────────────────────────
+
+async def test_a_file_of_several_chunks_arrives_whole(tmp_path):
+    node = FakeNode()
+    data = secrets.token_bytes(2 * CHUNK + 12345)
+    entry = node.add("share", "big.bin", data)
+    target = tmp_path / "big.bin"
+    seen = []
+    assert await download(node.session(), entry, target, progress=seen.append) == len(data)
+    assert target.read_bytes() == data
+    assert sum(seen) == len(data)
+    assert not list(tmp_path.glob(f"*{PART_SUFFIX}*"))
+    assert node.leases == {}, "the lease is closed at the end"
+
+
+async def test_an_interrupted_download_resumes_from_its_last_whole_chunk(tmp_path):
+    node = FakeNode()
+    data = secrets.token_bytes(3 * CHUNK + 7)
+    entry = node.add("share", "big.bin", data)
+    target = tmp_path / "big.bin"
+    part = tmp_path / f"big.bin{PART_SUFFIX}"
+    # A chunk and a half on disk: only the whole chunk counts.
+    part.write_bytes(data[:CHUNK + CHUNK // 2])
+    (tmp_path / f"big.bin{PART_SUFFIX}.json").write_text(
+        json.dumps({"id": entry["id"], "size": entry["size"]}))
+    await download(node.session(), entry, target)
+    assert target.read_bytes() == data
+    assert node.chunk_requests == 3
+
+
+async def test_a_part_of_another_file_is_started_over(tmp_path):
+    node = FakeNode()
+    data = secrets.token_bytes(2 * CHUNK)
+    entry = node.add("share", "big.bin", data)
+    (tmp_path / f"big.bin{PART_SUFFIX}").write_bytes(b"x" * CHUNK)
+    (tmp_path / f"big.bin{PART_SUFFIX}.json").write_text(
+        json.dumps({"id": "0" * 64, "size": entry["size"]}))
+    await download(node.session(), entry, tmp_path / "big.bin")
+    assert (tmp_path / "big.bin").read_bytes() == data
+    assert node.chunk_requests == 2
+
+
+async def test_a_refused_chunk_fails_the_file_and_keeps_the_part(tmp_path):
+    node = FakeNode()
+    entry = node.add("share", "big.bin", secrets.token_bytes(3 * CHUNK))
+    node.fail_chunk = (entry["id"], 2)
+    with pytest.raises(TransferError) as e:
+        await download(node.session(), entry, tmp_path / "big.bin")
+    assert "chunk 2" in e.value.message and "stopped answering" in e.value.message
+    assert not (tmp_path / "big.bin").exists()
+    assert (tmp_path / f"big.bin{PART_SUFFIX}").stat().st_size == 2 * CHUNK
+    assert node.leases == {}
+
+
+async def test_a_file_that_appeared_meanwhile_is_not_replaced(tmp_path):
+    node = FakeNode()
+    entry = node.add("share", "a.txt", b"from the group")
+    (tmp_path / "a.txt").write_bytes(b"mine")
+    with pytest.raises(TransferError):
+        await download(node.session(), entry, tmp_path / "a.txt")
+    assert (tmp_path / "a.txt").read_bytes() == b"mine"
+    await download(node.session(), entry, tmp_path / "a.txt", overwrite=True)
+    assert (tmp_path / "a.txt").read_bytes() == b"from the group"
+
+
+async def test_a_queued_lease_waits_for_its_grant(tmp_path):
+    node = FakeNode()
+    node.queue_first = 1
+    entry = node.add("share", "a.txt", b"hello")
+    queued = []
+    await download(node.session(), entry, tmp_path / "a.txt", on_queued=queued.append)
+    assert queued == [1] and (tmp_path / "a.txt").read_bytes() == b"hello"
+
+
+async def test_an_interrupted_download_keeps_its_part_and_closes_its_lease(tmp_path):
+    import asyncio
+    node = FakeNode()
+    node.chunk_delay = 0.05
+    data = secrets.token_bytes(20 * CHUNK)
+    entry = node.add("share", "big.bin", data)
+    task = asyncio.ensure_future(download(node.session(), entry, tmp_path / "big.bin"))
+    while node.chunk_requests < 12:
+        await asyncio.sleep(0.01)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert node.leases == {}
+    assert not (tmp_path / "big.bin").exists()
+    kept = (tmp_path / f"big.bin{PART_SUFFIX}").stat().st_size
+    assert kept > 0 and kept % CHUNK == 0
+    node.chunk_delay = 0
+    await download(node.session(), entry, tmp_path / "big.bin")
+    assert (tmp_path / "big.bin").read_bytes() == data
+
+
+# ── Upload ───────────────────────────────────────────────────────────────────
+
+async def test_an_upload_of_several_chunks_is_stored_whole():
+    node = FakeNode()
+    data = secrets.token_bytes(5 * UPLOAD_CHUNK + 99)
+    stored = await upload(node.session(), _reader(data), len(data), "scan.pdf", "share")
+    assert stored == "scan.pdf"
+    assert node.stored("share/scan.pdf") == data
+    assert node.leases == {}
+
+
+async def test_an_interrupted_upload_resumes_where_the_node_stopped():
+    node = FakeNode()
+    data = secrets.token_bytes(4 * UPLOAD_CHUNK)
+    node.uploads[("share", "scan.pdf")] = {"next": 2, "data": data[:2 * UPLOAD_CHUNK]}
+    sent = []
+    await upload(node.session(), _reader(data), len(data), "scan.pdf", "share",
+                 progress=sent.append)
+    assert node.stored("share/scan.pdf") == data
+    assert sum(sent) == len(data)
+
+
+async def test_a_name_taken_meanwhile_is_reported():
+    node = FakeNode()
+    node.add("share", "scan.pdf", b"theirs")
+    stored = await upload(node.session(), _reader(b"mine"), 4, "scan.pdf", "share")
+    assert stored == "scan (1).pdf"
+    assert node.stored("share/scan.pdf") == b"theirs"
+
+
+async def test_a_full_disk_is_fatal():
+    node = FakeNode()
+    node.refuse_upload_code = "disk_full"
+    with pytest.raises(TransferError) as e:
+        await upload(node.session(), _reader(b"x"), 1, "a.txt", "share")
+    assert e.value.fatal
+
+
+async def test_folders_are_created_where_asked():
+    node = FakeNode()
+    assert await create_dir(node.session(), "share", "Scans") == "share/Scans"
+    assert "share/Scans" in node.dirs
+
+
+# ── Names ────────────────────────────────────────────────────────────────────
+
+NAMES = [
+    "photo.jpg", "Été 2024", "東京", "track (1).flac", "[live] set", "a_b-c",
+    "rock & roll", "l’été", "x+y#z@w", "9lives",
+    ".DS_Store", "_private", "-dash", " space", "trailing ", "trailing.",
+    "a/b", "a\\b", "semi;colon", "x" * 128, "x" * 129, "", "tab\tname",
+]
+
+
+def test_the_cli_refuses_exactly_the_names_the_node_refuses():
+    roots = pytest.importorskip("meshbay_node.roots")
+    differ = [n for n in NAMES
+              if bool(roots.SAFE_UPLOAD_NAME.match(n)) != (upload_name_problem(n, is_dir=True)
+                                                           is None)]
+    assert not differ, f"the CLI and the node disagree on: {differ}"
+    for n in ["desktop.ini", "x.LNK", "a.url", "b.scf", "c.library-ms", "d.searchConnector-ms"]:
+        assert roots.shell_active(n) and upload_name_problem(n), n
+
+
+def test_a_downloaded_name_is_made_writable_and_stays_inside(tmp_path):
+    path, renamed = local_path(tmp_path, ["evil\x1b[31mred.txt"])
+    assert renamed and path.parent == tmp_path and "\x1b" not in path.name
+    path, renamed = local_path(tmp_path, ["CON.txt"])
+    assert renamed and path.name != "CON.txt"
+    path, renamed = local_path(tmp_path, ["été", "plage de l'ouest.jpg"])
+    assert not renamed and path == tmp_path / "été" / "plage de l'ouest.jpg"
+    path, _ = local_path(tmp_path, [".."])
+    assert path.parent == tmp_path
+
+
+def test_a_link_out_of_the_destination_is_refused(tmp_path):
+    outside = tmp_path / "outside"
+    outside.mkdir()
+    dest = tmp_path / "dest"
+    dest.mkdir()
+    (dest / "sub").symlink_to(outside)
+    with pytest.raises(UnsafeName):
+        local_path(dest, ["sub", "x.txt"])
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
new file mode 100644
index 0000000..4283819
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -0,0 +1,570 @@
+"""
+The real `meshbay-cli` against a real hub and a real node, both on loopback.
+
+Opt-in (MESHBAY_CLI_E2E=1): it starts a throwaway hub (SQLite, a fresh key) and
+a node in a home of its own, which takes a while and runs Argon2 at 128 MB
+several times. Nothing reaches a real hub or the developer's node.
+
+What it sets up, as an operator and a browser would:
+- an account, activated directly in the throwaway hub's database (no mail);
+- a group whose name holds a `/` and an `@`, which the hub allows and the path
+  syntax has to carry;
+- a node linked to the account, hosting the group over a folder whose names
+  are the awkward ones real data has: spaces, accents, an apostrophe, a long
+  name without spaces, and a name with an escape character in it;
+- the account's first contact with the node — a member invitation code, then a
+  keypair bundle left on the node — done here the way a browser does it,
+  because the CLI does not join groups.
+
+Then the CLI: login, ls of groups, of a root, recursive, JSON, a second run
+from the identity it kept, and logout.
+
+    MESHBAY_CLI_E2E=1 pytest packages/meshbay-cli/tests/test_e2e_local.py -v
+
+MESHBAY_CLI_E2E_BIN names another meshbay-cli to run in place of the one in
+this venv: a packaged build, through a wrapper that mounts it where it installs.
+"""
+
+import asyncio
+import base64
+import json
+import os
+import re
+import secrets
+import socket
+import sqlite3
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import httpx
+import pytest
+
+pytestmark = [
+    pytest.mark.skipif(os.environ.get("MESHBAY_CLI_E2E") != "1",
+                       reason="opt-in: set MESHBAY_CLI_E2E=1 (see the module docstring)"),
+    pytest.mark.timeout(600),
+]
+
+USERNAME = "alice-e2e-cli"
+PASSPHRASE = "a long enough test passphrase, nobody's"
+GROUP_NAME = "Some Saga/Box @Set"
+BIN = Path(sys.executable).parent
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+def _wait_http(url: str, timeout: float = 60) -> None:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        try:
+            httpx.get(url, timeout=2)
+            return
+        except httpx.HTTPError:
+            time.sleep(0.3)
+    raise AssertionError(f"{url} did not come up")
+
+
+@pytest.fixture(scope="module")
+def world(tmp_path_factory):
+    root = tmp_path_factory.mktemp("e2e")
+    procs: list[subprocess.Popen] = []
+    try:
+        yield _build(root, procs)
+    finally:
+        for p in procs:
+            p.terminate()
+        for p in procs:
+            try:
+                p.wait(10)
+            except subprocess.TimeoutExpired:
+                p.kill()
+
+
+def _build(root: Path, procs: list) -> dict:
+    from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from meshbay_cli.client.keys import auth_key
+
+    # ── The hub ──────────────────────────────────────────────────────────────
+    hub_home = root / "hub-home"
+    hub_home.mkdir()
+    key = root / "hub_private.pem"
+    key.write_bytes(Ed25519PrivateKey.generate().private_bytes(
+        serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption()))
+    db = root / "hub.db"
+    port = _free_port()
+    hub_url = f"http://127.0.0.1:{port}"
+    hub_env = {**os.environ, "HOME": str(hub_home),
+               "MESHBAY_DATABASE_URL": f"sqlite+aiosqlite:///{db}",
+               "MESHBAY_HUB_KEY": str(key), "MESHBAY_HUB_ID": "e2e-hub",
+               "MESHBAY_CAPTCHA_ENABLED": "false"}
+    hub_log = open(root / "hub.log", "w")
+    procs.append(subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "meshbay_hub.app:create_app", "--factory",
+         "--host", "127.0.0.1", "--port", str(port)],
+        env=hub_env, stdout=hub_log, stderr=subprocess.STDOUT))
+    _wait_http(f"{hub_url}/v1/hub/version")
+
+    ak = auth_key(USERNAME, PASSPHRASE)
+    with httpx.Client(base_url=hub_url, timeout=60) as hub:
+        r = hub.post("/v1/users/register", json={
+            "username": USERNAME, "email": "alice@example.invalid", "auth_key": ak})
+        assert r.status_code == 201, r.text
+        with sqlite3.connect(db) as conn:
+            conn.execute("UPDATE users SET status='active' WHERE username=?", (USERNAME,))
+        r = hub.post("/v1/users/login", json={"username": USERNAME, "auth_key": ak})
+        assert r.status_code == 200, r.text
+        auth = {"Authorization": f"Bearer {r.json()['access_token']}"}
+        r = hub.post("/v1/groups", headers=auth, json={
+            "name": GROUP_NAME, "visibility": "private", "join_policy": "invite"})
+        assert r.status_code == 201, r.text
+        group_id = r.json()["group_id"]
+
+        # ── The node ─────────────────────────────────────────────────────────
+        node_home = root / "node-home"
+        node_home.mkdir()
+        share = root / "share"
+        (share / "2024" / "été à la mer").mkdir(parents=True)
+        (share / "2024" / "été à la mer" / "plage de l'ouest.jpg").write_bytes(
+            secrets.token_bytes(3000))
+        (share / "2024" / "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac")\
+            .write_bytes(secrets.token_bytes(2_500_000))
+        (share / "notes.txt").write_text("hello\n")
+        (share / "evil\x1b[31mred.txt").write_text("escape\n")
+        node_env = {**os.environ, "HOME": str(node_home),
+                    "XDG_CONFIG_HOME": str(node_home / ".config"),
+                    "XDG_DATA_HOME": str(node_home / ".local/share"),
+                    "XDG_STATE_HOME": str(node_home / ".local/state")}
+        node = [str(BIN / "meshbay-node")]
+
+        def run_node(*args: str) -> str:
+            p = subprocess.run(node + list(args), env=node_env, capture_output=True,
+                               text=True, timeout=120)
+            assert p.returncode == 0, f"meshbay-node {args}: {p.stdout}\n{p.stderr}"
+            return p.stdout
+
+        out = run_node("init", "--hub-url", hub_url, "--username", USERNAME)
+        node_pk = re.search(r"Node key: (\S+)", out).group(1)
+        cfg = next(node_home.rglob("node.toml"))
+        ui_port = _free_port()
+        cfg.write_text(cfg.read_text().replace("ui_port   = 18000",
+                                               f"ui_port   = {ui_port}"))
+        r = hub.put("/v1/users/me/node_key", headers=auth,
+                    json={"pk_node_ed25519": node_pk})
+        assert r.status_code == 200, r.text
+
+        node_log = open(root / "node.log", "w")
+        procs.append(subprocess.Popen(node, env=node_env, stdout=node_log,
+                                      stderr=subprocess.STDOUT))
+        _wait_http(f"http://127.0.0.1:{ui_port}/", timeout=90)
+
+        # Attaching reloads the node by itself; the key needs the group loaded.
+        run_node("group", "add", group_id, "--dir", str(share))
+        deadline = time.monotonic() + 90
+        while "GEK ready" not in (out := _try(run_node, "gek", "init", "--group", group_id)):
+            assert time.monotonic() < deadline, f"gek init never succeeded: {out}"
+            time.sleep(2)
+        code = re.search(r"INVITATION CODE\s+(\S+)",
+                         run_node("member", "invite", USERNAME, "--group", group_id)).group(1)
+
+        # The node is online at the hub once it lists itself for the group.
+        deadline = time.monotonic() + 90
+        while not hub.get(f"/v1/groups/{group_id}/nodes", headers=auth).json()["nodes"]:
+            assert time.monotonic() < deadline, "the node never reached the hub"
+            time.sleep(1)
+
+    app = App(hub_url, group_id)
+    asyncio.run(app.first_contact(code))
+    return {"root": root, "hub_url": hub_url, "group_id": group_id, "db": db,
+            "share": share, "app": app}
+
+
+def _try(fn, *args) -> str:
+    try:
+        return fn(*args)
+    except AssertionError as e:
+        return str(e)
+
+
+class App:
+    """The application's side of the test: a device already linked to the node."""
+
+    def __init__(self, hub_url: str, group_id: str):
+        self.hub_url, self.group_id = hub_url, group_id
+        self.ident = None
+
+    async def session(self, code: str = ""):
+        """Signed in, handshake done, joined: (channel, close, node_pk, nonce_s, user_id)."""
+        import os as _os
+
+        from aiortc import (
+            RTCConfiguration,
+            RTCIceServer,
+            RTCPeerConnection,
+            RTCSessionDescription,
+        )
+        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+        from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+        from meshbay_cli.client.account import Account
+        from meshbay_cli.client.hub import Hub
+        from meshbay_cli.client.keys import Identity, b64, unb64
+        from meshbay_cli.client.session import STUN_SERVERS, Channel, _dtls_fingerprint
+        from meshbay_common import MNP_VERSION
+        from meshbay_common.crypto import unwrap_gek_aes
+        from meshbay_common.handshake import MNP_MIN_SUPPORTED, make_proof, webrtc_binding
+        from meshbay_common.join import join_transcript
+
+        hub = Hub(self.hub_url)
+        account = Account(hub, USERNAME, lambda: PASSPHRASE)
+        await account.sign_in(remember=False)
+        token = await account.token()
+        node = (await hub.group_nodes(token, self.group_id))[0]
+        mnp = await hub.mnp_token(token, node["pk_node"], self.group_id)
+        pc = RTCPeerConnection(RTCConfiguration(
+            iceServers=[RTCIceServer(urls=u) for u in STUN_SERVERS]))
+        dc = pc.createDataChannel("mnp", ordered=True)
+        ch = Channel(dc)
+        opened = asyncio.Event()
+        dc.on("open", opened.set)
+        await pc.setLocalDescription(await pc.createOffer())
+        offer = pc.localDescription.sdp
+        answer = await hub.offer(token, node["node_id"], offer)
+        await pc.setRemoteDescription(RTCSessionDescription(sdp=answer, type="answer"))
+        await asyncio.wait_for(opened.wait(), 30)
+        binding = webrtc_binding(_dtls_fingerprint(offer), _dtls_fingerprint(answer))
+        nonce_c = _os.urandom(32)
+        chal = await ch.request({"type": "handshake", "v": MNP_VERSION,
+                                 "v_min": MNP_MIN_SUPPORTED, "token": mnp,
+                                 "group_id": self.group_id, "nonce": b64(nonce_c)},
+                                "handshake_challenge")
+        nonce_s, node_pk = unb64(chal["nonce"]), chal["node_pk"]
+        if self.ident is None:
+            self.ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
+        ident, ts = self.ident, int(time.time())
+        joined = await ch.request({
+            "type": "join_request", "v": MNP_VERSION, "group_id": self.group_id,
+            "pk_ed25519": ident.pk_ed, "pk_x25519": ident.pk_x, "code": code, "ts": ts,
+            "sig": b64(ident.sk_ed.sign(join_transcript(
+                node_pk_b64=node_pk, group_id=self.group_id, user_id=account.user_id,
+                pk_ed25519_b64=ident.pk_ed, pk_x25519_b64=ident.pk_x,
+                nonce_node=nonce_s, ts=ts)))}, "join_result")
+        assert joined.get("ok"), joined
+        gek = unwrap_gek_aes(joined, ident.sk_x.private_bytes_raw(),
+                             ident.sk_x.public_key().public_bytes_raw())
+        await ch.request({"type": "handshake_response", "v": MNP_VERSION,
+                          "proof": b64(make_proof(gek, "client", self.group_id, nonce_c,
+                                                  nonce_s, binding))}, "handshake_ack")
+
+        async def close():
+            await pc.close()
+            await account.close()
+            await hub.aclose()
+        return ch, close, node_pk, nonce_s, account
+
+    async def first_contact(self, code: str) -> None:
+        """What a browser does the first time: join with the code, leave a bundle."""
+        import os as _os
+
+        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+        from meshbay_cli.client.keys import node_bundle_key
+        from meshbay_common import MNP_VERSION
+
+        ch, close, node_pk, _, account = await self.session(code)
+        # MBK3, as keyderive.js encryptBundle writes it.
+        nonce = _os.urandom(12)
+        aad = f"meshbay:bundle:v3|{account.user_id}|{node_pk}".encode()
+        ct = AESGCM(node_bundle_key(account.master, node_pk)).encrypt(
+            nonce, json.dumps(self.ident.to_pkcs8()).encode(), aad)
+        bundle = base64.b64encode(b"MBK3" + bytes([1]) + nonce + ct).decode()
+        stored = await ch.request({"type": "keypair_bundle_store", "v": MNP_VERSION,
+                                   "bundle_enc": bundle}, "ack")
+        assert stored.get("detail") == "keypair_bundle_stored", stored
+        await close()
+
+    async def withdraw_bundle(self) -> None:
+        """Browser access turned off: the application withdraws the node's copy."""
+        from meshbay_common import MNP_VERSION
+        ch, close, *_ = await self.session()
+        r = await ch.request({"type": "keypair_bundle_delete", "v": MNP_VERSION}, "ack")
+        assert r.get("detail") == "keypair_bundle_deleted", r
+        await close()
+
+    async def approve(self, code: str) -> None:
+        """Approve a device by the code it shows, as the group's Settings tab does."""
+        from meshbay_cli.client.keys import b64
+        from meshbay_common import MNP_VERSION
+        from meshbay_common.device import device_add_transcript, device_code_hash
+        ch, close, node_pk, nonce_s, account = await self.session()
+        listed = await ch.request({"type": "device_lookup", "v": MNP_VERSION},
+                                  "device_lookup_result")
+        raw = code.replace("-", "").upper()
+        match = next(r for r in listed["requests"]
+                     if device_code_hash(raw, r["pk_ed25519"], r["pk_x25519"])
+                     == r["code_hash"])
+        ts = int(time.time())
+        sig = self.ident.sk_ed.sign(device_add_transcript(
+            node_pk_b64=node_pk, user_id=account.user_id,
+            pk_ed25519_b64=match["pk_ed25519"], pk_x25519_b64=match["pk_x25519"],
+            nonce_node=nonce_s, ts=ts))
+        await ch.request({"type": "device_add", "v": MNP_VERSION,
+                          "pk_ed25519": match["pk_ed25519"],
+                          "pk_x25519": match["pk_x25519"],
+                          "code_hash": match["code_hash"], "ts": ts, "sig": b64(sig)},
+                         "device_add_ack")
+        await close()
+
+
+def cli(world: dict, *args: str, stdin: str | None = None,
+        home: str = "cli") -> subprocess.CompletedProcess:
+    env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / home),
+           "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
+    env.pop("MESHBAY_PASSWORD", None)
+    exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
+    return subprocess.run([exe, *args], env=env, input=stdin,
+                          capture_output=True, text=True, timeout=180)
+
+
+def _handle() -> str:
+    return GROUP_NAME.replace("%", "%25").replace("/", "%2F") + "@" + USERNAME
+
+
+def test_the_cli_against_a_real_hub_and_node(world):
+    hub = world["hub_url"]
+
+    r = cli(world, "--hub", hub, "-u", USERNAME, "login", "--password-file", "-",
+            stdin=PASSPHRASE + "\n")
+    assert r.returncode == 0, r.stderr
+    assert "Signed in" in r.stdout and "1 group" in r.stdout
+    # No key storage in this environment: the file fallback, said once.
+    assert "No OS key storage here" in r.stderr
+
+    # The groups, from the hub alone; the stored session needs no passphrase.
+    r = cli(world, "ls")
+    assert r.returncode == 0, r.stderr
+    assert r.stdout.strip() == _handle()
+    r = cli(world, "groups")
+    assert r.returncode == 0, r.stderr
+    print(r.stdout)
+    [line] = r.stdout.splitlines()[1:]
+    assert line.split()[:3] == ["up", "owner", "private"] and line.endswith(_handle())
+
+    # One group, by its handle: its root.
+    r = cli(world, "-vv", "ls", "-l", _handle())
+    assert r.returncode == 0, r.stderr
+    assert "keypair_bundle_fetch" in r.stderr     # first contact: the node's copy
+    assert re.search(r"^root\s.*share/\s+\(writable\)$", r.stdout, re.M), r.stdout
+
+    # Recursive: the awkward names arrive, the escape character is escaped.
+    r = cli(world, "-vv", "ls", "-R", "-l", f"{_handle()}/share")
+    assert r.returncode == 0, r.stderr
+    assert "keypair_bundle_fetch" not in r.stderr  # the identity was kept
+    out = r.stdout
+    assert "plage de l'ouest.jpg" in out
+    assert "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac" in out
+    assert "\x1b" not in out and "evil\\x1b[31mred.txt" in out
+    assert "été à la mer/" in out
+
+    # By name alone, case-insensitively, and as JSON.
+    r = cli(world, "--json", "ls", "some saga%2Fbox @set/share/2024")
+    assert r.returncode == 0, r.stderr
+    items = json.loads(r.stdout)
+    names = {i["name"]: i for i in items}
+    assert names["été à la mer"]["kind"] == "dir"
+    flac = names["A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac"]
+    assert flac["kind"] == "file" and flac["type"] == "audio"
+    assert flac["size"] == 2_500_000 and len(flac["id"]) == 64
+
+    # Not found is exit 3; an unknown group too.
+    assert cli(world, "ls", f"{_handle()}/share/nope").returncode == 3
+    assert cli(world, "ls", "No Such Group").returncode == 3
+
+    # ── get ──────────────────────────────────────────────────────────────────
+    out_dir = world["root"] / "downloads"
+    flac = "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac"
+    r = cli(world, "get", f"{_handle()}/share/2024/{flac}", "-o", f"{out_dir}/")
+    assert r.returncode == 0, r.stderr
+    assert (out_dir / flac).read_bytes() == (world["share"] / "2024" / flac).read_bytes()
+    r = cli(world, "get", "-R", f"{_handle()}/share", "-o", f"{out_dir}/tree/")
+    assert r.returncode == 0, r.stderr
+    got = out_dir / "tree" / "share"
+    assert (got / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes() == \
+        (world["share"] / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes()
+    # The escape character could not be written as it is: renamed, and said so.
+    assert "evil\\x1b[31mred.txt is saved as evil_[31mred.txt" in r.stderr
+    assert (got / "evil_[31mred.txt").exists()
+    r = cli(world, "get", f"{_handle()}/share/notes.txt", "-o", "-")
+    assert r.returncode == 0 and r.stdout == "hello\n"
+
+    # ── put ──────────────────────────────────────────────────────────────────
+    local = world["root"] / "to-send"
+    (local / "Scans" / "2024").mkdir(parents=True)
+    scan = secrets.token_bytes(300_000)
+    (local / "Scans" / "2024" / "page 1.pdf").write_bytes(scan)
+    (local / "Scans" / "empty").mkdir()
+    (local / "Notes.TXT").write_bytes(b"clash")
+    r = cli(world, "put", "-R", str(local / "Scans"), f"{_handle()}/share/2024")
+    assert r.returncode == 0, r.stderr
+    assert (world["share"] / "2024" / "Scans" / "2024" / "page 1.pdf").read_bytes() == scan
+    assert (world["share"] / "2024" / "Scans" / "empty").is_dir()
+    # notes.txt is there already, whatever the case: nothing sent.
+    r = cli(world, "put", str(local / "Notes.TXT"), f"{_handle()}/share")
+    assert r.returncode == 8 and "already there" in r.stderr
+    assert not (world["share"] / "Notes.TXT").exists()
+    r = cli(world, "put", "-", f"{_handle()}/share/piped.txt", stdin="through stdin")
+    assert r.returncode == 0, r.stderr
+    assert (world["share"] / "piped.txt").read_text() == "through stdin"
+    # And what was sent can be fetched back, once the node has indexed it.
+    deadline = time.monotonic() + 60
+    while True:
+        r = cli(world, "get", f"{_handle()}/share/2024/Scans/2024/page 1.pdf", "-o",
+                str(out_dir / "back.pdf"))
+        if r.returncode == 0 or time.monotonic() > deadline:
+            break
+        time.sleep(1)
+    assert r.returncode == 0, r.stderr
+    assert (out_dir / "back.pdf").read_bytes() == scan
+
+    # ── ls -R over every group ───────────────────────────────────────────────
+    deadline = time.monotonic() + 60
+    while True:
+        r = cli(world, "ls", "-R", "/")
+        assert r.returncode == 0, r.stderr
+        if "piped.txt" in r.stdout or time.monotonic() > deadline:
+            break
+        time.sleep(1)       # the node indexes what it received on its own schedule
+    assert "page 1.pdf" in r.stdout and "piped.txt" in r.stdout, r.stdout
+
+    # The device key was registered, and logout retires it.
+    with sqlite3.connect(world["db"]) as conn:
+        assert conn.execute("SELECT count(*) FROM user_devices").fetchone()[0] == 1
+    r = cli(world, "logout")
+    assert r.returncode == 0, r.stderr
+    assert "Signed out." in r.stdout
+    with sqlite3.connect(world["db"]) as conn:
+        assert conn.execute("SELECT count(*) FROM user_devices").fetchone()[0] == 0
+    assert not (world["root"] / "cli" / "secrets.enc").exists()
+
+    # Signed out: a run needs credentials again, and says so.
+    r = cli(world, "ls")
+    assert r.returncode == 2 and "Not signed in" in r.stderr
+
+
+def test_a_changed_node_key_is_refused_until_reset(world):
+    def run(*args, stdin=None):
+        return cli(world, *args, stdin=stdin, home="cli-trust")
+
+    r = run("--hub", world["hub_url"], "-u", USERNAME, "login", "--password-file", "-",
+            stdin=PASSPHRASE + "\n")
+    assert r.returncode == 0, r.stderr
+    assert run("ls", _handle()).returncode == 0
+
+    r = run("--json", "trust", "list")
+    [(node_id, pk)] = json.loads(r.stdout).items()
+    state_path = world["root"] / "cli-trust" / "state.json"
+    state = json.loads(state_path.read_text())
+    state["pins"][node_id] = base64.b64encode(b"\x01" * 32).decode()
+    state_path.write_text(json.dumps(state))
+    r = run("ls", _handle())
+    assert r.returncode == 6 and "trust reset" in r.stderr, r.stderr
+    assert run("trust", "reset", "no-such-node").returncode == 3
+    r = run("trust", "reset", node_id)
+    assert r.returncode == 0, r.stderr
+    assert run("ls", _handle()).returncode == 0
+    assert json.loads(run("--json", "trust", "list").stdout) == {node_id: pk}
+    assert run("logout").returncode == 0
+
+
+def test_a_machine_with_no_identity_there_is_linked_as_a_device(world):
+    """Browser access off: the node keeps no copy, so a new machine asks to be
+    approved, shows a code, and joins with its own keys once approved."""
+    app = world["app"]
+    asyncio.run(app.withdraw_bundle())
+
+    def run(*args, stdin=None):
+        return cli(world, *args, stdin=stdin, home="cli2")
+
+    r = run("--hub", world["hub_url"], "-u", USERNAME, "login", "--password-file", "-",
+            stdin=PASSPHRASE + "\n")
+    assert r.returncode == 0, r.stderr
+
+    # No terminal here: the code is shown, and the run ends rather than waits.
+    r = run("ls", _handle())
+    assert r.returncode == 5, r.stderr
+    found = re.search(r"enter\s+([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr)
+    assert found, r.stderr
+    code = found.group(1)
+    # A second run shows the same code: the request it filed is still pending.
+    r = run("ls", _handle())
+    assert r.returncode == 5 and code in r.stderr, r.stderr
+
+    asyncio.run(app.approve(code))
+    r = run("ls", "-l", _handle())
+    assert r.returncode == 0, r.stderr
+    assert "share/" in r.stdout
+
+    assert run("logout").returncode == 0
+
+
+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."""
+    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.screens import MainScreen
+
+    # Browser access off, whichever test ran before: this machine must be linked.
+    await world["app"].withdraw_bundle()
+    r = await asyncio.to_thread(cli, world, "--hub", world["hub_url"], "-u", USERNAME,
+                                "login", "--password-file", "-",
+                                stdin=PASSPHRASE + "\n", home="cli-tui")
+    assert r.returncode == 0, r.stderr
+    monkeypatch.setenv("MESHBAY_CLI_HOME", str(world["root"] / "cli-tui"))
+    monkeypatch.chdir(tmp_path)
+    args = main_module.build_parser().parse_args(["--hub", world["hub_url"]])
+    app = MeshBayApp(Context(args))
+
+    def names(pane):
+        return [pane.table.get_row_at(i)[0].plain.split("  (")[0].strip()
+                for i in range(pane.table.row_count)]
+
+    async def until(pilot, cond, timeout=60):
+        deadline = time.monotonic() + timeout
+        while not cond():
+            shown = (names(app.screen.query_one(RemotePane))
+                     if isinstance(app.screen, MainScreen) else None)
+            assert time.monotonic() < deadline, (
+                f"timed out on {type(app.screen).__name__}; "
+                f"notified: {[n.message for n in app._notifications]}; remote: {shown}")
+            await pilot.pause(0.1)
+
+    async with app.run_test(size=(140, 40)) as pilot:
+        await until(pilot, lambda: isinstance(app.screen, MainScreen))
+        remote = app.screen.query_one(RemotePane)
+        name = _handle() + "/"
+        await until(pilot, lambda: name in names(remote))
+        remote.table.move_cursor(row=names(remote).index(name))
+        remote.table.focus()
+        await pilot.press("enter")
+        # The node keeps no copy of the identity: this machine is a new device,
+        # shows its code, and waits while it is approved elsewhere — then the
+        # group opens by itself.
+        await until(pilot, lambda: any("enter " in n.message for n in app._notifications))
+        note = next(n.message for n in app._notifications if "enter " in n.message)
+        code = re.search(r"enter ([0-9A-Z]{4}-[0-9A-Z]{4})", note).group(1)
+        await world["app"].approve(code)
+        await until(pilot, lambda: "share/" in names(remote))
+        remote.table.move_cursor(row=names(remote).index("share/"))
+        await pilot.press("enter")
+        await until(pilot, lambda: "notes.txt" in names(remote))
+        remote.table.move_cursor(row=names(remote).index("notes.txt"))
+        await pilot.press("f5")
+        await until(pilot, lambda: (tmp_path / "notes.txt").exists()
+                    and (tmp_path / "notes.txt").read_text() == "hello\n")
diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py
new file mode 100644
index 0000000..680b9e0
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_interactive.py
@@ -0,0 +1,159 @@
+"""
+The interactive mode, driven by Textual's pilot against the in-memory hub and
+node: what a person does with the keys, and what lands on disk and on the node.
+"""
+
+import asyncio
+import time
+
+import pytest
+from fake_hub import PASSPHRASE, USERNAME, FakeHub
+from fake_node import FakeNode
+from meshbay_cli import __main__ as cli
+from meshbay_cli.client import pool
+from meshbay_cli.commands import context
+from meshbay_cli.tui.app import MeshBayApp
+from meshbay_cli.tui.panes import LocalPane, RemotePane
+from meshbay_cli.tui.screens import MainScreen, SignInScreen
+from meshbay_cli.tui.widgets import Choice
+
+
+@pytest.fixture
+def world(monkeypatch, tmp_path):
+    hub, node = FakeHub(), FakeNode()
+    monkeypatch.setattr(context, "Hub", lambda url: hub)
+
+    async def open_group(account, group):
+        return node.session()
+    monkeypatch.setattr(pool, "open_group", open_group)
+    work = tmp_path / "work"
+    work.mkdir()
+    monkeypatch.chdir(work)
+    return hub, node, work
+
+
+def _app() -> MeshBayApp:
+    args = cli.build_parser().parse_args(["--hub", "https://hub.test"])
+    return MeshBayApp(context.Context(args))
+
+
+async def _until(pilot, condition, timeout: float = 10) -> None:
+    deadline = time.monotonic() + timeout
+    while not condition():
+        assert time.monotonic() < deadline, "timed out"
+        await pilot.pause(0.05)
+
+
+def _names(pane) -> list[str]:
+    """Row names as shown, without the mark or a note such as "(read-only)"."""
+    return [pane.table.get_row_at(i)[0].plain.split("  (")[0].strip()
+            for i in range(pane.table.row_count)]
+
+
+async def _open(pilot, pane, name: str) -> None:
+    await _until(pilot, lambda: name in _names(pane))
+    pane.table.move_cursor(row=_names(pane).index(name))
+    pane.table.focus()
+    await pilot.press("enter")
+
+
+async def test_signing_in_then_downloading_and_uploading(world):
+    hub, node, work = world
+    node.add("share", "notes.txt", b"hello")
+    node.add("share/2024", "a.jpg", b"A" * 3000)
+    node.add("share", "[bold]x[/bold]\x1b.txt", b"tricky")
+    (work / "mine.txt").write_bytes(b"from here")
+
+    app = _app()
+    async with app.run_test(size=(140, 40)) as pilot:
+        # Nothing stored: the sign-in screen, then the panes.
+        await _until(pilot, lambda: isinstance(app.screen, SignInScreen))
+        app.screen.query_one("#user").value = USERNAME
+        app.screen.query_one("#pass").value = PASSPHRASE
+        app.screen.query_one("#pass").focus()
+        await pilot.press("enter")
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        assert len(hub.devices) == 1, "signing in here remembers the machine"
+
+        remote = app.screen.query_one(RemotePane)
+        local = app.screen.query_one(LocalPane)
+        await _open(pilot, remote, f"Family@{USERNAME}/")
+        await _open(pilot, remote, "share/")
+        await _until(pilot, lambda: "notes.txt" in _names(remote))
+
+        # A name somebody else chose is shown as text, escapes and all.
+        assert "[bold]x[/bold]\\x1b.txt" in _names(remote)
+
+        # Mark two, F5: both land in the local folder.
+        for name in ("notes.txt", "2024/"):
+            remote.table.move_cursor(row=_names(remote).index(name))
+            await pilot.press("space")
+        await pilot.press("f5")
+        await _until(pilot, lambda: (work / "2024" / "a.jpg").exists()
+                     and (work / "notes.txt").exists())
+        assert (work / "notes.txt").read_bytes() == b"hello"
+        assert (work / "2024" / "a.jpg").read_bytes() == b"A" * 3000
+
+        # Again: already there, asked once for the batch.
+        remote.table.move_cursor(row=_names(remote).index("notes.txt"))
+        remote.table.focus()
+        (work / "notes.txt").write_bytes(b"changed here")
+        await pilot.press("f5")
+        await _until(pilot, lambda: isinstance(app.screen, Choice))
+        await pilot.click("#choice-skip")
+        await pilot.pause(0.2)
+        assert (work / "notes.txt").read_bytes() == b"changed here"
+
+        # Upload from the left into the open folder.
+        await _until(pilot, lambda: isinstance(app.screen, MainScreen))
+        local.action_refresh()
+        await _until(pilot, lambda: "mine.txt" in _names(local))
+        local.table.move_cursor(row=_names(local).index("mine.txt"))
+        local.table.focus()
+        await pilot.press("f5")
+        await _until(pilot, lambda: node.stored("share/mine.txt") == b"from here")
+        # The node's index changed: the right pane shows it.
+        await _until(pilot, lambda: "mine.txt" in _names(remote))
+
+        # A name already there is asked about, never replaced.
+        local.table.focus()
+        await pilot.press("f5")
+        await _until(pilot, lambda: isinstance(app.screen, Choice))
+        await pilot.click("#choice-cancel")
+        await pilot.pause(0.2)
+        assert node.stored("share/mine.txt") == b"from here"
+
+
+async def test_a_remembered_session_opens_on_the_panes(world, capsys):
+    hub, node, _ = world
+    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 _until(pilot, lambda: f"Family@{USERNAME}/" in _names(remote))
+
+
+async def test_a_read_only_folder_takes_no_upload(world):
+    hub, node, work = world
+    node.roots[0]["writable"] = False
+    node.add("share", "notes.txt", b"hello")
+    (work / "mine.txt").write_bytes(b"x")
+    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)
+        local = app.screen.query_one(LocalPane)
+        await _open(pilot, remote, f"Family@{USERNAME}/")
+        await _open(pilot, remote, "share/")
+        await _until(pilot, lambda: "notes.txt" in _names(remote))
+        await _until(pilot, lambda: "mine.txt" in _names(local))
+        local.table.move_cursor(row=_names(local).index("mine.txt"))
+        local.table.focus()
+        await pilot.press("f5")
+        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)
diff --git a/packages/meshbay-cli/tests/test_listing.py b/packages/meshbay-cli/tests/test_listing.py
new file mode 100644
index 0000000..407074e
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_listing.py
@@ -0,0 +1,73 @@
+"""The index as the Files tab draws it, and names printed safely."""
+
+import pytest
+from meshbay_cli.client.tree import Tree
+from meshbay_cli.errors import NOT_FOUND, CliError
+from meshbay_cli.output import human_size, safe
+
+INDEX = {
+    "roots": [{"name": "share", "available": True, "writable": True},
+              {"name": "usb", "available": False, "writable": False}],
+    "dirs": ["share", "share/2024", "share/empty", "usb", "usb/old"],
+    "entries": [
+        {"id": "a" * 64, "name": "notes.txt", "path": "share", "size": 6, "type": "document"},
+        {"id": "b" * 64, "name": "x.flac", "path": "share/2024", "size": 9, "type": "audio"},
+        # A folder the node's walk did not list still exists through its file.
+        {"id": "c" * 64, "name": "y.jpg", "path": "share/2024/late", "size": 1,
+         "type": "image"},
+        {"id": "d" * 64, "name": "z.txt", "path": "usb/old", "size": 1, "type": "document"},
+    ],
+}
+
+
+def _names(items):
+    return sorted(i.name for i in items)
+
+
+def test_the_top_is_the_roots():
+    tree = Tree.from_index(INDEX)
+    top = tree.children("")
+    assert _names(top) == ["share", "usb"]
+    assert {i.name: i.available for i in top} == {"share": True, "usb": False}
+
+
+def test_a_folder_holds_its_folders_and_files():
+    tree = Tree.from_index(INDEX)
+    assert _names(tree.children("share")) == ["2024", "empty", "notes.txt"]
+    assert _names(tree.children("share/2024")) == ["late", "x.flac"]
+
+
+def test_an_unavailable_root_shows_nothing_unless_asked():
+    tree = Tree.from_index(INDEX)
+    assert tree.children("usb") == []
+    assert _names(tree.children("usb/old", include_unavailable=True)) == ["z.txt"]
+
+
+def test_walk_visits_every_folder():
+    tree = Tree.from_index(INDEX)
+    assert [p for p, _ in tree.walk("share")] == [
+        "share", "share/2024", "share/2024/late", "share/empty"]
+
+
+def test_a_missing_folder_is_not_found():
+    with pytest.raises(CliError) as e:
+        Tree.from_index(INDEX).children("share/nope")
+    assert e.value.code == NOT_FOUND
+
+
+def test_a_file_is_found_by_its_path():
+    tree = Tree.from_index(INDEX)
+    assert tree.file("share/2024/x.flac")["id"] == "b" * 64
+    assert tree.file("share/2024/nope") is None
+
+
+def test_control_characters_and_bidi_overrides_are_escaped():
+    assert safe("evil\x1b[31mred") == "evil\\x1b[31mred"
+    assert safe("a\nb\x7fc\x9b") == "a\\x0ab\\x7fc\\x9b"
+    assert safe("photo‮gpj.exe") == "photo\\u202egpj.exe"
+    assert safe("été 🎵 l'ouest") == "été 🎵 l'ouest"
+
+
+def test_sizes():
+    assert human_size(6) == "6 B"
+    assert human_size(2_500_000) == "2.4 MB"
diff --git a/packages/meshbay-cli/tests/test_manpage.py b/packages/meshbay-cli/tests/test_manpage.py
new file mode 100644
index 0000000..ebd71c8
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_manpage.py
@@ -0,0 +1,70 @@
+"""
+The man page is generated from the parser; these keep the generator honest.
+
+An option added to an action must reach the page, every exit code must be
+explained, and troff must read the result without a warning.
+"""
+
+import argparse
+import os
+import shutil
+import subprocess
+
+import pytest
+from meshbay_cli import __main__ as cli
+from meshbay_cli import errors, manpage
+
+
+def _section(page: str, start: str) -> str:
+    return page.split(start + "\n", 1)[1].split("\n.S", 1)[0]
+
+
+def _entry(option: str) -> str:
+    """How an option opens its entry: `.TP`, then the option in bold."""
+    return f"\\fB{manpage._roff(option)}\\fR"
+
+
+def test_every_action_and_option_is_in_its_section():
+    page = manpage.render()
+    parser = cli.build_parser()
+    glob = _section(page, ".SH GLOBAL OPTIONS")
+    for a in parser._actions:
+        for option in a.option_strings:
+            assert _entry(option) in glob, f"global {option} missing"
+    sub = next(a for a in parser._actions if isinstance(a, argparse._SubParsersAction))
+    assert list(sub.choices) == list(cli.ACTIONS)
+    for name, sp in sub.choices.items():
+        own = _section(page, f".SS {name}")
+        for a in sp._actions:
+            if isinstance(a, argparse._HelpAction):
+                continue
+            if a.option_strings and _entry(a.option_strings[0]) in glob:
+                continue                               # a global one, listed once
+            if a.option_strings:
+                for option in a.option_strings:
+                    assert _entry(option) in own, f"{name} {option} missing"
+            elif a.choices:
+                for choice in a.choices:
+                    assert _entry(choice) in own, f"{name} {choice} missing"
+            else:
+                meta = a.metavar or a.dest.upper()
+                assert f"\\fI{manpage._roff(meta)}\\fR" in own, f"{name} {meta} missing"
+
+
+def test_every_exit_code_is_explained():
+    codes = {v for k, v in vars(errors).items() if k.isupper() and isinstance(v, int)}
+    assert codes | {0} == set(manpage.EXIT_CODES)
+
+
+def test_names_and_text_cannot_break_the_page():
+    assert manpage._roff(".hidden") == "\\&.hidden"
+    assert manpage._roff("a\\b — c-d") == "a\\eb \\[u2014] c\\-d"
+
+
+@pytest.mark.skipif(not shutil.which("groff"), reason="groff not installed")
+def test_troff_reads_it_without_a_warning(tmp_path):
+    page = tmp_path / "meshbay-cli.1"
+    page.write_text(manpage.render(), encoding="ascii")    # nothing past ASCII left
+    r = subprocess.run(["groff", "-man", "-Tutf8", "-ww", "-z", str(page)],
+                       capture_output=True, text=True, env={**os.environ, "LC_ALL": "C"})
+    assert r.returncode == 0 and not r.stderr, r.stderr
diff --git a/packages/meshbay-cli/tests/test_remote_paths.py b/packages/meshbay-cli/tests/test_remote_paths.py
new file mode 100644
index 0000000..5192e7b
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_remote_paths.py
@@ -0,0 +1,77 @@
+"""Remote paths and group handles — `docs/MESHBAY_DESIGN.md` §8.4, `group-link.js`."""
+
+import pytest
+from meshbay_cli.client.paths import encode_name, handle, parse, resolve_group
+from meshbay_cli.errors import NOT_FOUND, USAGE, CliError
+
+GROUPS = [
+    {"id": "1", "name": "Family", "owner_username": "alice"},
+    {"id": "2", "name": "Family", "owner_username": "bob"},
+    {"id": "3", "name": "Some Saga/Box @Set", "owner_username": "alice"},
+    {"id": "4", "name": "me@home", "owner_username": "carol"},
+    {"id": "5", "name": "Work", "owner_username": "Dave"},
+]
+
+
+def test_nothing_or_a_slash_is_every_group():
+    assert parse("") is None
+    assert parse("/") is None
+
+
+def test_a_typed_path_decodes_only_the_slash_and_the_percent():
+    p = parse("Some Saga%2FBox @Set@alice/share/100% done.txt")
+    assert p.group == "Some Saga/Box @Set@alice"
+    assert p.parts == ["share", "100% done.txt"]
+    assert parse("G/a%252Fb").parts == ["a%2Fb"]
+
+
+def test_a_link_copied_from_the_application_is_decoded_in_full():
+    p = parse("https://hub.example/#/Some%20Saga%2FBox%20@Set@alice/share/b%C3%A9b%C3%A9.jpg")
+    assert p.from_link
+    assert p.group == "Some Saga/Box @Set@alice"
+    assert p.parts == ["share", "bébé.jpg"]
+
+
+def test_dot_segments_are_refused():
+    with pytest.raises(CliError) as e:
+        parse("Family/share/../etc")
+    assert e.value.code == USAGE
+
+
+def test_names_are_nfc():
+    decomposed = "été"
+    assert parse(f"G/{decomposed}").parts == ["été"]
+
+
+def test_the_owner_is_after_the_last_at():
+    assert resolve_group(GROUPS, "Family@bob")["id"] == "2"
+    assert resolve_group(GROUPS, "Some Saga/Box @Set@alice")["id"] == "3"
+    assert resolve_group(GROUPS, "me@home@carol")["id"] == "4"
+
+
+def test_a_name_with_an_at_of_its_own_still_resolves_without_owner():
+    assert resolve_group(GROUPS, "me@home")["id"] == "4"
+
+
+def test_names_and_owners_compare_without_case():
+    assert resolve_group(GROUPS, "some saga/box @set")["id"] == "3"
+    assert resolve_group(GROUPS, "work@dave")["id"] == "5"
+
+
+def test_two_groups_of_one_name_ask_for_the_owner():
+    with pytest.raises(CliError) as e:
+        resolve_group(GROUPS, "Family")
+    assert e.value.code == USAGE
+    assert "Family@alice" in e.value.message and "Family@bob" in e.value.message
+
+
+def test_an_unknown_group_is_not_found():
+    with pytest.raises(CliError) as e:
+        resolve_group(GROUPS, "Nope")
+    assert e.value.code == NOT_FOUND
+
+
+def test_a_handle_can_be_typed_back():
+    for g in GROUPS:
+        assert resolve_group(GROUPS, parse(handle(g)).group)["id"] == g["id"]
+    assert encode_name("100%/x") == "100%25%2Fx"
diff --git a/packages/meshbay-cli/tests/test_selftest.py b/packages/meshbay-cli/tests/test_selftest.py
new file mode 100644
index 0000000..ea49fa1
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_selftest.py
@@ -0,0 +1,16 @@
+"""The frozen build's self-test, run here from source so it cannot rot unseen."""
+
+from meshbay_cli import __main__ as cli
+
+
+def test_the_self_test_passes_from_source(monkeypatch, tmp_path, capsys):
+    monkeypatch.setenv("MESHBAY_CLI_SELFTEST", "1")
+    monkeypatch.setenv("MESHBAY_CLI_HOME", str(tmp_path))    # restored after
+    assert cli.main([]) == 0
+    assert "self-test ok" in capsys.readouterr().out
+
+
+def test_it_is_not_reachable_with_arguments(monkeypatch, capsys):
+    monkeypatch.setenv("MESHBAY_CLI_SELFTEST", "1")
+    assert cli.main(["--version"]) == 0
+    assert "self-test" not in capsys.readouterr().out
diff --git a/packages/meshbay-cli/tests/test_store_and_session.py b/packages/meshbay-cli/tests/test_store_and_session.py
new file mode 100644
index 0000000..42458e2
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_store_and_session.py
@@ -0,0 +1,276 @@
+"""
+The local store, and the hub session it keeps.
+
+The case that matters most is two runs renewing at once: the hub rotates refresh
+tokens and treats a spent one as theft, so a second renewal with the token the
+first has just spent would sign the account out everywhere.
+"""
+
+import asyncio
+import os
+import sys
+import time
+
+import keyring
+import pytest
+from fake_hub import PASSPHRASE, USERNAME, FakeHub
+from keyring.backend import KeyringBackend
+from meshbay_cli.client.account import Account, no_passphrase
+from meshbay_cli.client.store import KEY_FILE, KEY_OS, Store
+from meshbay_cli.errors import AUTH, CliError
+
+
+class MemoryKeyring(KeyringBackend):
+    priority = 1
+
+    def __init__(self):
+        super().__init__()
+        self.items: dict[tuple[str, str], str] = {}
+
+    def get_password(self, service, username):
+        return self.items.get((service, username))
+
+    def set_password(self, service, username, password):
+        self.items[(service, username)] = password
+
+    def delete_password(self, service, username):
+        self.items.pop((service, username), None)
+
+
+@pytest.fixture
+def memory_keyring():
+    backend = MemoryKeyring()
+    previous = keyring.get_keyring()
+    keyring.set_keyring(backend)
+    yield backend
+    keyring.set_keyring(previous)
+
+
+def _saved(root, secrets=None, state=None) -> Store:
+    store = Store(root)
+    store.state = {"hub": "https://hub.test", "username": USERNAME, "user_id": "u",
+                   **(state or {})}
+    store.secrets = {"access_token": "old", "access_exp": 0, "refresh_token": "rt0",
+                     **(secrets or {})}
+    store.save(new=True)
+    return store
+
+
+# ── The store ────────────────────────────────────────────────────────────────
+
+def test_without_key_storage_the_key_is_a_private_file(tmp_path):
+    store = _saved(tmp_path / "s")
+    assert store.state["key_store"] == KEY_FILE
+    if sys.platform != "win32":
+        for name in ("store.key", "secrets.enc", "state.json"):
+            assert os.stat(tmp_path / "s" / name).st_mode & 0o077 == 0, name
+    again = Store(tmp_path / "s")
+    assert again.load() and again.secrets["refresh_token"] == "rt0"
+
+
+def test_with_key_storage_the_key_is_kept_there(tmp_path, memory_keyring):
+    a = _saved(tmp_path / "a")
+    _saved(tmp_path / "b")
+    assert a.state["key_store"] == KEY_OS and not (tmp_path / "a" / "store.key").exists()
+    # One entry per store: a second store never overwrites the first's key.
+    assert len(memory_keyring.items) == 2
+    assert Store(tmp_path / "a").load() and Store(tmp_path / "b").load()
+    a.erase()
+    assert len(memory_keyring.items) == 1
+    assert not Store(tmp_path / "a").exists()
+
+
+def test_secrets_are_not_readable_in_the_file(tmp_path):
+    _saved(tmp_path / "s", secrets={"refresh_token": "a-very-recognisable-token"})
+    assert b"a-very-recognisable-token" not in (tmp_path / "s" / "secrets.enc").read_bytes()
+
+
+def test_a_store_whose_key_storage_vanished_says_so(tmp_path, memory_keyring):
+    _saved(tmp_path / "s")
+    keyring.set_keyring(__import__("keyring.backends.fail", fromlist=["Keyring"]).Keyring())
+    with pytest.raises(CliError) as e:
+        Store(tmp_path / "s").load()
+    assert e.value.code == AUTH and "OS key storage" in e.value.message
+
+
+def test_peek_reads_the_state_without_the_key(tmp_path):
+    _saved(tmp_path / "s")
+    (tmp_path / "s" / "store.key").unlink()
+    assert Store(tmp_path / "s").peek()["username"] == USERNAME
+
+
+# ── Renewal ──────────────────────────────────────────────────────────────────
+
+def _open(root, hub) -> Account:
+    return Account.open(hub, None, no_passphrase, Store(root))
+
+
+async def test_two_runs_renewing_at_once_spend_the_token_once(tmp_path):
+    hub = FakeHub()
+    first = hub._session()
+    _saved(tmp_path / "s", secrets={"refresh_token": first["refresh_token"]})
+    hub.refresh_delay = 0.2      # wide enough for the second run to arrive meanwhile
+
+    a, b = _open(tmp_path / "s", hub), _open(tmp_path / "s", hub)
+    ta, tb = await asyncio.gather(a.token(), b.token())
+
+    assert not hub.reuse_detected
+    assert hub.refresh_calls == 1
+    assert ta == tb and ta in hub.access
+    # And the stored token is the live one: a third run renews without incident.
+    c = _open(tmp_path / "s", hub)
+    c.secrets["access_exp"] = 0
+    await c.token(rejected=c.secrets["access_token"])
+    assert not hub.reuse_detected
+
+
+async def test_a_token_still_fresh_is_used_as_it_is(tmp_path):
+    hub = FakeHub()
+    s = hub._session()
+    _saved(tmp_path / "s", secrets={"access_token": s["access_token"],
+                                    "access_exp": time.time() + 3600,
+                                    "refresh_token": s["refresh_token"]})
+    assert await _open(tmp_path / "s", hub).token() == s["access_token"]
+    assert hub.refresh_calls == 0
+
+
+async def test_a_refused_refresh_falls_back_to_the_device_key(tmp_path):
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from meshbay_cli.client.keys import b64
+    hub = FakeHub()
+    sk = Ed25519PrivateKey.generate()
+    hub.devices["dev-1"] = b64(sk.public_key().public_bytes_raw())
+    _saved(tmp_path / "s", secrets={"refresh_token": "revoked",
+                                    "device_sk": b64(sk.private_bytes_raw())})
+    token = await _open(tmp_path / "s", hub).token()
+    assert token in hub.access
+
+
+async def test_a_session_that_ended_asks_for_login(tmp_path):
+    hub = FakeHub()
+    _saved(tmp_path / "s", secrets={"refresh_token": "revoked"})
+    with pytest.raises(CliError) as e:
+        await _open(tmp_path / "s", hub).token()
+    assert e.value.code == AUTH and "meshbay-cli login" in e.value.message
+
+
+async def test_a_401_renews_once_and_replays(tmp_path):
+    hub = FakeHub()
+    s = hub._session()
+    _saved(tmp_path / "s", secrets={"access_token": "revoked-by-the-hub",
+                                    "access_exp": time.time() + 3600,
+                                    "refresh_token": s["refresh_token"]})
+    groups = await _open(tmp_path / "s", hub).call(hub.my_groups)
+    assert groups[0]["name"] == "Family" and hub.refresh_calls == 1
+
+
+# ── Signing in ───────────────────────────────────────────────────────────────
+
+async def test_sign_in_derives_the_keys_and_registers_one_device(tmp_path):
+    hub = FakeHub()
+    acc = Account(hub, USERNAME, lambda: PASSPHRASE)
+    await acc.sign_in(remember=True)
+    assert acc.master and acc.legacy and acc.user_id
+    assert len(hub.devices) == 1 and acc.state["device_id"] == "dev-1"
+    # Signing in again keeps the same device key rather than adding one.
+    await acc.sign_in(remember=True)
+    assert len(hub.devices) == 1
+
+
+async def test_a_wrong_passphrase_is_an_auth_error():
+    with pytest.raises(CliError) as e:
+        await Account(FakeHub(), USERNAME, lambda: "wrong").sign_in(remember=False)
+    assert e.value.code == AUTH
+
+
+async def test_a_renewal_by_passphrase_is_what_gets_stored(tmp_path):
+    """The fallback replaces the session wholesale; the store must get the new one."""
+    hub = FakeHub()
+    _saved(tmp_path / "s", secrets={"refresh_token": "revoked"})
+    acc = Account.open(hub, None, lambda: PASSPHRASE, Store(tmp_path / "s"))
+    await acc.token()
+    again = Store(tmp_path / "s")
+    again.load()
+    assert again.secrets["refresh_token"] == acc.secrets["refresh_token"]
+    assert again.secrets["refresh_token"] in hub.refresh_tokens
+    assert again.secrets.get("master")
+
+
+async def test_a_run_signed_out_meanwhile_says_so(tmp_path):
+    hub = FakeHub()
+    _saved(tmp_path / "s")
+    acc = _open(tmp_path / "s", hub)
+    Store(tmp_path / "s").erase()
+    with pytest.raises(CliError) as e:
+        await acc.token()
+    assert e.value.code == AUTH and "another run" in e.value.message
+
+
+async def test_a_pending_device_request_is_kept_then_adopted(tmp_path):
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+    from meshbay_cli.client.keys import Identity
+    hub = FakeHub()
+    _saved(tmp_path / "s")
+    acc = _open(tmp_path / "s", hub)
+    ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
+    await acc.keep_pending("node-pk", ident, "ABCD-EFGH", time.time() + 600)
+
+    # A later run shows the same code and joins with the same keys.
+    again = _open(tmp_path / "s", hub)
+    pending = again.pending_identity("node-pk")
+    assert pending.code == "ABCD-EFGH" and pending.identity.pk_ed == ident.pk_ed
+    assert again.identity("node-pk") is None
+    await again.adopt_pending("node-pk")
+    assert again.pending_identity("node-pk") is None
+    assert _open(tmp_path / "s", hub).identity("node-pk").pk_ed == ident.pk_ed
+
+
+async def test_an_expired_device_request_is_not_reused(tmp_path):
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+    from meshbay_cli.client.keys import Identity
+    hub = FakeHub()
+    _saved(tmp_path / "s")
+    acc = _open(tmp_path / "s", hub)
+    ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
+    await acc.keep_pending("node-pk", ident, "ABCD-EFGH", time.time() - 1)
+    assert _open(tmp_path / "s", hub).pending_identity("node-pk") is None
+
+
+async def test_a_machine_waiting_for_approval_retries_until_it_is_approved(monkeypatch):
+    from meshbay_cli.client import groups
+    from meshbay_cli.client.session import DeviceLinkPending
+    monkeypatch.setattr(groups, "LINK_POLL", 0)
+    tries, shown = [], []
+
+    async def connect(account, group_id, node):
+        tries.append(node["node_id"])
+        if len(tries) < 3:
+            raise DeviceLinkPending("ABCD-EFGH", time.time() + 600)
+        return "session"
+    monkeypatch.setattr(groups, "connect", connect)
+
+    class Acc:
+        def on_device_code(self, group, pending, *, first):
+            shown.append((pending.link_code, first))
+            return True
+    got = await groups._connect_linking(Acc(), {"id": "g"}, {"node_id": "n1"})
+    assert got == "session" and tries == ["n1", "n1", "n1"]
+    assert shown == [("ABCD-EFGH", True), ("ABCD-EFGH", False)]
+
+
+async def test_without_a_terminal_the_code_is_shown_and_the_run_ends(monkeypatch):
+    from meshbay_cli.client import groups
+    from meshbay_cli.client.session import DeviceLinkPending
+
+    async def connect(account, group_id, node):
+        raise DeviceLinkPending("ABCD-EFGH", time.time() + 600)
+    monkeypatch.setattr(groups, "connect", connect)
+
+    class Acc:
+        def on_device_code(self, group, pending, *, first):
+            return False
+    with pytest.raises(DeviceLinkPending) as e:
+        await groups._connect_linking(Acc(), {"id": "g"}, {"node_id": "n1"})
+    assert e.value.code == 5 and e.value.link_code == "ABCD-EFGH"