feat(cli): chat --follow

chat GROUP --follow prints the last messages, then each one the node pushes,
until Ctrl+C (exit 130). The connection is pinged every half minute; one
that closes, fails or stops answering is opened again, and what was said
meanwhile is read from the history. Each message is printed once, by its
(device, nonce) pair, which the node stores once. With --json, one object
per line. The man page lists it, with an example, and its FILES now names
the device keys kept for the Tier 2 check.

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

3 files changed, 128 insertions, 0 deletions

diff --git a/packages/meshbay-cli/tests/fake_node.py b/packages/meshbay-cli/tests/fake_node.py
index 8f4356b..b4e64ff 100644
--- a/packages/meshbay-cli/tests/fake_node.py
+++ b/packages/meshbay-cli/tests/fake_node.py
@@ -76,6 +76,7 @@ class FakeNode:
             "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.unresponsive = False            # pings go unanswered, as a dead link's
         self.admin_ops: dict[str, dict] = {}
 
     def add(self, path: str, name: str, data: bytes, uploader_id: str = "") -> dict:
@@ -273,6 +274,11 @@ class FakeChannel:
         return {"type": "chat_keys_resp", "group_id": GROUP_ID,
                 **seal(n.gek, PURPOSE_CHAT_KEYS, "chat_keys_resp", GROUP_ID, payload)}
 
+    def _ping(self, msg):
+        if self.node.unresponsive:
+            return {"type": "error", "detail": "no answer"}
+        return {"type": "pong", "token": msg.get("token")}
+
     def _group_roster_req(self, msg):
         n = self.node
         return {"type": "group_roster_resp", "group_id": GROUP_ID,
diff --git a/packages/meshbay-cli/tests/test_chat_follow.py b/packages/meshbay-cli/tests/test_chat_follow.py
new file mode 100644
index 0000000..cc6db99
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_chat_follow.py
@@ -0,0 +1,94 @@
+"""
+`chat --follow`: new messages as the node pushes them, a dead connection opened
+again, and what was said meanwhile read from the history — each message once.
+"""
+
+import asyncio
+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.commands import chat, context
+
+
+@pytest.fixture
+def follow(monkeypatch, capsys):
+    hub, node = FakeHub(), FakeNode()
+    monkeypatch.setattr(context, "Hub", lambda url: hub)
+    sessions = []
+
+    async def open_group(account, group):
+        sessions.append(node.session())
+        return sessions[-1]
+    monkeypatch.setattr(chat, "open_group", open_group)
+    monkeypatch.setattr(chat, "PING_EVERY", 0.05)
+    monkeypatch.setattr(chat, "RECONNECT_WAITS", (0.01,))
+    assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
+                     "login"]) == 0
+    capsys.readouterr()
+
+    def start(*argv):
+        args = cli.build_parser().parse_args(cli.chat.rewrite_count(list(argv), cli.ACTIONS))
+        return asyncio.ensure_future(chat.run(context.Context(args)))
+    return node, sessions, start
+
+
+def _push(session, node):
+    session.ch._emit({**node.chat[-1], "type": "chat_msg"})
+
+
+async def _until(cond, timeout=5.0):
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not cond():
+        assert asyncio.get_running_loop().time() < deadline, "timed out"
+        await asyncio.sleep(0.02)
+
+
+async def test_follow_prints_pushes_reconnects_and_never_repeats(follow, capsys):
+    node, sessions, start = follow
+    node.say("before")
+    task = start("--json", "chat", "-1", "Family", "--follow")
+    printed = []
+
+    def lines():
+        printed.extend(json.loads(x)["text"] for x in capsys.readouterr().out.splitlines())
+        return printed
+    try:
+        await _until(lambda: lines() == ["before"])
+
+        node.say("live one")
+        _push(sessions[-1], node)
+        await _until(lambda: lines()[-1:] == ["live one"])
+
+        # The connection drops; meanwhile two messages are stored and only the
+        # last of them is pushed, once the new connection is up.
+        sessions[-1].ch.closed.set()
+        node.say("while away")
+        await _until(lambda: len(sessions) == 2)
+        await _until(lambda: "while away" in lines())
+        node.say("after")
+        _push(sessions[-1], node)
+        _push(sessions[-1], node)                     # the node said it twice
+        await _until(lambda: lines()[-1:] == ["after"])
+
+        # A connection that stops answering its ping is replaced too.
+        node.unresponsive = True
+        await _until(lambda: len(sessions) == 3)
+        node.unresponsive = False
+        node.say("on the third connection")
+        _push(sessions[-1], node)
+        await _until(lambda: lines()[-1:] == ["on the third connection"])
+        await asyncio.sleep(0.2)
+        assert lines() == ["before", "live one", "while away", "after",
+                           "on the third connection"]
+    finally:
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+
+def test_follow_and_post_are_not_asked_at_once(follow, capsys):
+    assert cli.main(["chat", "Family", "-f", "-m", "x"]) == 1
+    assert cli.main(["chat", "-f"]) == 1
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py
index cf6a3b9..359ab06 100644
--- a/packages/meshbay-cli/tests/test_e2e_local.py
+++ b/packages/meshbay-cli/tests/test_e2e_local.py
@@ -1068,3 +1068,31 @@ async def test_the_interactive_chat_against_the_real_node(world, monkeypatch, tm
 
     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
+
+
+def test_chat_follow_against_the_real_node(world):
+    """`chat --follow` in its own process: messages posted from another run
+    arrive as the node pushes them, across more than one liveness ping, and
+    Ctrl+C ends it. Runs on the machine the chat tests above linked."""
+    import signal
+    group = _handle()
+    env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / "cli-chat"),
+           "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"}
+    exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli")
+    follower = subprocess.Popen([exe, "--json", "chat", "-1", group, "--follow"], env=env,
+                                stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    try:
+        first = json.loads(follower.stdout.readline())          # the last one, at once
+        assert first["text"]
+        for text, wait in (("followed one", 0), ("followed two, after a ping", 36)):
+            time.sleep(wait)
+            r = cli(world, "chat", group, "-m", text, home="cli-chat")
+            assert r.returncode == 0, r.stderr
+            got = json.loads(follower.stdout.readline())
+            assert got["text"] == text and got["sender"] == USERNAME, got
+            assert got["trust"] in ("pinned", "first", "linked")
+    finally:
+        follower.send_signal(signal.SIGINT)
+        _, err = follower.communicate(timeout=20)
+    assert follower.returncode == 130
+    assert "reconnecting" not in err, err