feat(cli): chat — Tier 2 check on the signing device, interactive room

Each message's signing device is checked against the group's roster as the
page checks it (a port of transport-roster.js, held to it by the parity
test): chains of countersignatures walked from each account's roots, the
keys accepted per account and node kept in the store, the roster read again
once before a key is called changed. Only "changed" is said, under the
message; --json carries every verdict.

chat with no group opens the interactive chat: the groups, then one group's
conversation IRC-style with pushed messages arriving live and a line to
write in; Esc or /part goes back to the groups, /quit or Ctrl+Q quits. Like
the bare command, it opens on the sign-in screen when nothing is stored.

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

4 files changed, 277 insertions, 4 deletions

diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index f5fe631..8f4356b 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -32,6 +32,7 @@ from meshbay_common.device import (
 from meshbay_common.groupbox import (
     PURPOSE_CHAT_KEYS,
     PURPOSE_INDEX,
+    PURPOSE_ROSTER,
     PURPOSE_UPLOAD,
     seal,
     unseal,
@@ -69,6 +70,12 @@ class FakeNode:
         self.ack: dict = {"chat_epoch": 1}   # the sealed configuration, opened
         self.chat_epochs = {1: secrets.token_bytes(32)}
         self.chat: list[dict] = []           # stored rows, oldest first
+        # group_roster_resp's devices; this machine's own key is a root of its account.
+        self.roster: list[dict] = [{
+            "user_id": USER_ID, "username": "alice-test", "pk_ed25519": self.identity.pk_ed,
+            "pk_x25519": self.identity.pk_x, "added_by_pk": "", "add_sig": "",
+            "add_nonce": "", "add_ts": 0}]
+        self.members: dict[str, Identity] = {}   # sender → the device it signs with
         self.admin_ops: dict[str, dict] = {}
 
     def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
@@ -86,17 +93,47 @@ class FakeNode:
         return GroupSession(None, FakeChannel(self), GROUP_ID, "node-1", NODE_PK, self.gek,
                             dict(self.ack), self.identity, self.nonce, self.role)
 
+    def list_device(self, sender: str, device: Identity, *,
+                    countersigned_by: Identity | None = None, proof: bool = True) -> None:
+        """A roster row for `sender`'s device: a root, or countersigned by another
+        of its devices — with the signature, or (`proof=False`) naming the
+        countersigner and carrying nothing, as a lying node would."""
+        row = {"user_id": "u-" + sender, "username": sender, "pk_ed25519": device.pk_ed,
+               "pk_x25519": device.pk_x, "added_by_pk": "", "add_sig": "",
+               "add_nonce": "", "add_ts": 0}
+        if countersigned_by is not None:
+            row["added_by_pk"] = countersigned_by.pk_ed
+            if proof:
+                nonce, ts = secrets.token_bytes(32), 1_700_000_000
+                row.update(add_nonce=base64.b64encode(nonce).decode(), add_ts=ts,
+                           add_sig=base64.b64encode(countersigned_by.sk_ed.sign(
+                               device_add_transcript(
+                                   node_pk_b64=NODE_PK, user_id="u-" + sender,
+                                   pk_ed25519_b64=device.pk_ed,
+                                   pk_x25519_b64=device.pk_x,
+                                   nonce_node=nonce, ts=ts))).decode())
+        self.roster.append(row)
+
     def say(self, text: str, *, sender: str = "bob-test", epoch: int = 1,
-            forge: bool = False) -> None:
-        """Another member's message, sealed and signed as a client does; `forge`
-        signs it with a key other than the device it names."""
-        other = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
+            forge: bool = False, device: Identity | None = None) -> Identity:
+        """Another member's message, sealed and signed as a client does, from
+        `device` (default: the sender's usual one, listed in the roster as a
+        root the first time); `forge` signs it with a key other than the device
+        it names."""
+        if device is None:
+            if sender not in self.members:
+                self.members[sender] = Identity(Ed25519PrivateKey.generate(),
+                                                X25519PrivateKey.generate())
+                self.list_device(sender, self.members[sender])
+            device = self.members[sender]
+        other = device
         sealed = chatbox.seal(self.chat_epochs[epoch], GROUP_ID, epoch, other.pk_ed,
                               base64.b64decode(other.pk_ed), other.sk_ed,
                               {"text": text, "sender_name": sender})
         if forge:
             sealed["sig"] = Ed25519PrivateKey.generate().sign(b"something else")
         self._store(base64.b64decode(other.pk_ed), epoch, sealed, "u-" + sender)
+        return other
 
     def _store(self, device: bytes, epoch: int, sealed: dict, sender_id: str) -> None:
         self.chat.append({"id": len(self.chat) + 1, "sender_id": sender_id,
@@ -236,6 +273,12 @@ class FakeChannel:
         return {"type": "chat_keys_resp", "group_id": GROUP_ID,
                 **seal(n.gek, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID, payload)}
 
+    def _group_roster_req(self, msg):
+        n = self.node
+        return {"type": "group_roster_resp", "group_id": GROUP_ID,
+                **seal(n.gek, PURPOSE_ROSTER, "group_roster_resp", GROUP_ID,
+                       {"devices": n.roster, "node_pk": NODE_PK})}
+
     def _chat_hist(self, msg):
         rows = self.node.chat
         if msg.get("before"):
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
index 19f5ca9..cf6a3b9 100644
--- a/packages/meshbay-cli/tests/test_e2e_local.py
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -1017,3 +1017,54 @@ def test_chat_between_the_command_line_and_the_application(world):
             break
         assert time.monotonic() < deadline, (r.stdout, r.stderr)
         time.sleep(1)
+
+
+async def test_the_interactive_chat_against_the_real_node(world, monkeypatch, tmp_path):
+    """The room open on a real group: a message posted from another run arrives
+    live, and one typed in the room is read back by `chat`. Runs after the
+    chat test above, on the machine it linked."""
+    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.chat import ChatGroupsScreen, ChatRoomScreen
+    from textual.widgets import DataTable, Input, RichLog
+
+    group = _handle()
+    r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat")
+    assert r.returncode == 0, "the chat test above did not leave this machine linked"
+    monkeypatch.setenv("MESHBAY_CLI_HOME", str(world["root"] / "cli-chat"))
+    app = MeshBayApp(Context(main_module.build_parser().parse_args(
+        ["--hub", world["hub_url"]])), ChatGroupsScreen)
+
+    def text(screen) -> str:
+        return "\n".join("".join(s.text for s in strip)
+                         for strip in screen.query_one(RichLog).lines)
+
+    async def until(pilot, cond, timeout=60):
+        deadline = time.monotonic() + timeout
+        while not cond():
+            assert time.monotonic() < deadline, [n.message for n in app._notifications]
+            await pilot.pause(0.1)
+
+    async with app.run_test(size=(140, 40)) as pilot:
+        await until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen)
+                    and app.screen.query_one(DataTable).row_count >= 1)
+        await pilot.press("enter")
+        await until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
+        room = app.screen
+        await until(pilot, lambda: "hello from the browser" in text(room))
+        assert "cannot be read" not in text(room) and "not seen before" not in text(room)
+
+        r = await asyncio.to_thread(cli, world, "chat", group, "-m",
+                                    "live from another run", home="cli-chat")
+        assert r.returncode == 0, r.stderr
+        await until(pilot, lambda: f"<{USERNAME}> live from another run" in text(room))
+
+        room.query_one(Input).value = "typed in the room"
+        await pilot.press("enter")
+        await until(pilot, lambda: f"<{USERNAME}> typed in the room" in text(room))
+        await pilot.press("escape")
+        await until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
+
+    r = await asyncio.to_thread(cli, world, "chat", "-1", group, home="cli-chat")
+    assert r.stdout.strip().endswith(f"{USERNAME}: typed in the room"), r.stdout
diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py
index 49180d5..fa7e96c 100644
--- a/packages/meshbay-cli/tests/test_interactive.py
+++ b/packages/meshbay-cli/tests/test_interactive.py
@@ -206,3 +206,82 @@ async def test_delete_removes_what_is_yours_and_asks_first(world):
         await pilot.press("delete")
         await pilot.pause(0.2)
         assert any("select them on the right" in n.message for n in app._notifications)
+
+
+def _log_text(screen) -> str:
+    from textual.widgets import RichLog
+    return "\n".join("".join(seg.text for seg in strip)
+                     for strip in screen.query_one(RichLog).lines)
+
+
+async def test_the_interactive_chat_joins_talks_leaves_and_comes_back(world, monkeypatch):
+    from meshbay_cli.tui.chat import ChatGroupsScreen, ChatRoomScreen
+    from textual.widgets import DataTable, Input
+    hub, node, _ = world
+    hub.groups.append({"id": "g9", "name": "Quiet", "owner_username": "bob-test",
+                       "node_online": True, "is_admin": False})
+    quiet = FakeNode()
+    quiet.ack["enabled_apps"] = ["files"]
+    sessions = {}
+
+    async def open_group(account, group):
+        target = quiet if group["id"] == "g9" else node
+        sessions[group["id"]] = target.session()
+        return sessions[group["id"]]
+    monkeypatch.setattr(pool, "open_group", open_group)
+    node.say("earlier, from bob")
+    assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
+                                              "--password", PASSPHRASE, "login"]) == 0
+
+    app = MeshBayApp(context.Context(cli.build_parser().parse_args(
+        ["--hub", "https://hub.test"])), ChatGroupsScreen)
+    async with app.run_test(size=(120, 30)) as pilot:
+        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
+        table = app.screen.query_one(DataTable)
+        await _until(pilot, lambda: table.row_count == 2)
+
+        # A group whose chat is off is not entered.
+        table.move_cursor(row=1)                      # Quiet@bob-test
+        await pilot.press("enter")
+        await _until(pilot, lambda: any("chat turned off" in n.message
+                                        for n in app._notifications))
+        assert isinstance(app.screen, ChatGroupsScreen)
+
+        table.move_cursor(row=0)                      # Family@alice-test
+        await pilot.press("enter")
+        await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
+        room = app.screen
+        await _until(pilot, lambda: "<bob-test> earlier, from bob" in _log_text(room))
+
+        # Posting: sealed, signed, stored on the node, shown here at once.
+        room.query_one(Input).value = "hello from the room"
+        await pilot.press("enter")
+        await _until(pilot, lambda: "<alice-test> hello from the room" in _log_text(room))
+        assert len(node.chat) == 2
+
+        # What the node pushes appears as it arrives.
+        bob = node.say("live, from bob")
+        sessions["g1"].ch._emit({**node.chat[-1], "type": "chat_msg"})
+        await _until(pilot, lambda: "<bob-test> live, from bob" in _log_text(room))
+        assert bob.pk_ed
+
+        # /part goes back; entering again shows the conversation once, and a
+        # pushed message once — the room left behind no handler.
+        room.query_one(Input).value = "/part"
+        await pilot.press("enter")
+        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
+        await pilot.press("enter")
+        await _until(pilot, lambda: isinstance(app.screen, ChatRoomScreen))
+        room = app.screen
+        await _until(pilot, lambda: "live, from bob" in _log_text(room))
+        node.say("once")
+        sessions["g1"].ch._emit({**node.chat[-1], "type": "chat_msg"})
+        await _until(pilot, lambda: "<bob-test> once" in _log_text(room))
+        await pilot.pause(0.2)
+        assert _log_text(room).count("<bob-test> once") == 1
+
+        # Esc goes back too; q quits from the list.
+        await pilot.press("escape")
+        await _until(pilot, lambda: isinstance(app.screen, ChatGroupsScreen))
+        await pilot.press("q")
+    assert app.return_code in (0, None)
diff --git a/packages/meshbay-cli/tests/test_roster.py b/packages/meshbay-cli/tests/test_roster.py
new file mode 100644
index 0000000..cdaf90b
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_roster.py
@@ -0,0 +1,100 @@
+"""
+Tier 2 for chat (docs/MESHBAY_DESIGN.md §3.3): whose device signed a message,
+checked against the group's roster and the keys this machine accepted before —
+through `main()`, against the in-memory node, with real countersignatures.
+"""
+
+import json
+
+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.keys import Identity
+from meshbay_cli.client.roster import RosterCheck, judge, verify_roster
+from meshbay_cli.commands import chat, context
+
+
+def _new() -> Identity:
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+    return Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
+
+
+@pytest.fixture
+def node(monkeypatch, capsys):
+    hub, fake = FakeHub(), FakeNode()
+    monkeypatch.setattr(context, "Hub", lambda url: hub)
+
+    async def open_group(account, group):
+        return fake.session()
+    monkeypatch.setattr(chat, "open_group", open_group)
+    assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
+                     "login"]) == 0
+    capsys.readouterr()
+    return fake
+
+
+def _last(capsys, n=1) -> list[dict]:
+    assert cli.main(["--json", "chat", f"-{n}", "Family"]) == 0
+    return json.loads(capsys.readouterr().out)
+
+
+def test_each_verdict_on_a_message_signing_device(node, capsys):
+    phone = node.say("from bob's phone")
+    assert _last(capsys)[0]["trust"] == "first"            # first sight: pinned
+    assert _last(capsys)[0]["trust"] == "pinned"
+
+    # A second device, countersigned by the first: no notice.
+    laptop = _new()
+    node.list_device("bob-test", laptop, countersigned_by=phone)
+    node.say("from bob's laptop", device=laptop)
+    assert _last(capsys)[0]["trust"] == "linked"
+
+    # Named as countersigned and carrying no proof, as a lying node would list it.
+    planted = _new()
+    node.list_device("bob-test", planted, countersigned_by=phone, proof=False)
+    node.say("hello, it's bob", device=planted)
+    assert _last(capsys)[0]["trust"] == "changed"
+    # Not in the roster at all.
+    node.say("hello again", device=_new())
+    assert _last(capsys)[0]["trust"] == "changed"
+
+    # Said in plain text under the message, and nowhere else.
+    assert cli.main(["chat", "-5", "Family"]) == 0
+    out = capsys.readouterr().out.splitlines()
+    assert sum("key you have not seen before" in line for line in out) == 2
+    assert out[-1].lstrip().startswith("! This account")
+    assert out[-3].lstrip().startswith("! This account")
+
+    # Another account, first sight.
+    node.say("hi from carol", sender="carol-test")
+    assert _last(capsys)[0]["trust"] == "first"
+
+
+async def test_a_device_listed_after_the_roster_was_read_is_not_a_substitution(node):
+    """The roster is read once per connection; a member's new device added since
+    is read again before "changed" is said."""
+    from meshbay_cli.client.account import Account
+    from meshbay_cli.client.store import Store
+    store = Store()
+    account = Account.open(context.Hub("x"), None, lambda: PASSPHRASE, store)
+    session = node.session()
+    phone = node.say("one")
+    check = RosterCheck(account, session)
+    assert await check.status("u-bob-test", phone.pk_ed) == "first"
+    tablet = _new()
+    node.list_device("bob-test", tablet, countersigned_by=phone)   # after the read
+    assert await check.status("u-bob-test", tablet.pk_ed) == "linked"
+
+
+def test_a_root_is_a_device_naming_no_countersigner():
+    a, b = _new(), _new()
+    rows = [{"user_id": "u", "pk_ed25519": a.pk_ed, "pk_x25519": a.pk_x,
+             "added_by_pk": ""},
+            {"user_id": "u", "pk_ed25519": b.pk_ed, "pk_x25519": b.pk_x,
+             "added_by_pk": a.pk_ed, "add_sig": "", "add_nonce": "", "add_ts": 0}]
+    got = verify_roster({"devices": rows, "node_pk": "pk"}, "pk")["u"]
+    assert got.verified == [a.pk_ed] and got.unevidenced == [b.pk_ed]
+    assert judge([a.pk_ed], got, b.pk_ed) == ("changed", None)
+    assert judge(None, got, b.pk_ed)[0] == "first"