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>

9 files changed, 319 insertions, 22 deletions

diff --git a/CLAUDE.md b/CLAUDE.md
index 350e22c..57ebbd4 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -1168,7 +1168,7 @@ which pytest also runs when `ANDROID_HOME` is set (`test_android_shell.py`).
 | Remote paths, listing, local names | `client/paths.py`, `client/tree.py`, `client/names.py` | `names.UPLOAD_NAME` is copied from the node's `SAFE_UPLOAD_NAME` |
 | Downloads and uploads | `client/transfers.py`, `client/upload_plan.py` | §5.5; `upload_plan` is shared by `put` and the interactive mode |
 | Search across groups | `commands/search.py` | `search_listed` honoured before the index is asked; one result per content hash |
-| Chat: read and post | `client/chat.py`, `commands/chat.py` | `chatbox` seal/verify; signature before decryption; unreadable shown, never dropped |
+| Chat: read and post | `client/chat.py`, `commands/chat.py` | `chatbox` seal/verify; signature before decryption; unreadable shown, never dropped; `--follow` reconnects and prints each message once (`Follower`) |
 | Chat: whose device signed (Tier 2) | `client/roster.py` | port of `transport-roster.js`, held to it by `test_account_pinning.py`; account pins in `state.json` |
 | Interactive chat | `tui/chat.py` | `meshbay-cli chat` alone; pushed `chat_msg` live; Esc or /part back to the groups |
 | Deleting files | `client/remove.py`, `commands/rm.py` | §5.4: the uploader's account or the node's operator; the whole run is checked before asking |
diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md
index 561ac82..2b0a166 100644
--- a/docs/MESHBAY_DESIGN.md
+++ b/docs/MESHBAY_DESIGN.md
@@ -3061,7 +3061,10 @@ the signing device is checked against the group's roster as the application
 checks it (Tier 2, §3.3): 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 — and only "changed" is said, under
-the message. `chat` alone opens the interactive chat: the groups, one group's
+the message. `chat GROUP --follow` then prints what the node pushes, pings the
+connection every half minute, opens it again when it dies and reads what was
+said meanwhile from the history, each message printed once (a device and nonce
+pair is stored once). `chat` alone opens the interactive chat: the groups, one group's
 conversation IRC-style with what the node pushes arriving live, and back.
 
 **Deleting** (`rm`) is `file_delete` signed with the machine's key on the node
diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md
index ded4cb1..e8bd8fb 100644
--- a/docs/USERGUIDE.md
+++ b/docs/USERGUIDE.md
@@ -242,6 +242,7 @@ meshbay-cli ls -l Family/photos         # a group, a folder of it
 meshbay-cli search beach 2024           # every group at once; paths get takes
 meshbay-cli chat Family                 # the last 10 messages; -100 for 100
 meshbay-cli chat Family -m "On my way"  # post one (-m - reads it from stdin)
+meshbay-cli chat Family --follow        # then each new message, until Ctrl+C
 meshbay-cli chat                        # interactive: pick a group, talk, Esc to leave
 meshbay-cli get -R Family/photos/2024   # download, here; -o elsewhere
 meshbay-cli put -R ~/scans Family/documents
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/chat.py b/packages/meshbay-cli/src/meshbay_cli/client/chat.py
index 60629e2..22a99ef 100644
--- a/packages/meshbay-cli/src/meshbay_cli/client/chat.py
+++ b/packages/meshbay-cli/src/meshbay_cli/client/chat.py
@@ -48,6 +48,9 @@ class Message:
     text: str
     unreadable: str = ""   # why it could not be read: signature, epoch, ...
     trust: str = ""        # roster.RosterCheck's verdict on the signing device
+    # Which message this is, wherever it was read from — history or a push,
+    # which carries no id: the node stores a (device, nonce) pair only once.
+    key: str = ""
 
 
 class ChatKeys:
@@ -100,6 +103,7 @@ async def open_row(row: dict, keys: ChatKeys, group_id: str,
     if not (device and nonce and ct and sig):
         msg.unreadable = "envelope"
         return msg
+    msg.key = f"{b64(bytes(device))}:{b64(bytes(nonce))}"
     if not verify(group_id, epoch, bytes(device), bytes(nonce), bytes(ct), bytes(sig)):
         msg.unreadable = "signature"
         return msg
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/chat.py b/packages/meshbay-cli/src/meshbay_cli/commands/chat.py
index 6fa084d..d650af3 100644
--- a/packages/meshbay-cli/src/meshbay_cli/commands/chat.py
+++ b/packages/meshbay-cli/src/meshbay_cli/commands/chat.py
@@ -8,8 +8,17 @@ asks for another count. `chat GROUP -m TEXT` posts TEXT (`-m -` reads it from
 stdin), sealed and signed as the application seals and signs it. Each message
 is shown with its time and the name its sender sealed in it; one that cannot
 be read is shown as such, never left out.
+
+`chat GROUP --follow` prints the last messages, then each new one as the node
+pushes it, until Ctrl+C. A connection that drops — or that stops answering the
+ping sent every half minute, as a sleeping laptop's does — is opened again,
+and what was said meanwhile is read from the history: each message is printed
+once, whichever way it came (the node stores a device and nonce pair once).
+With `--json`, one object per line.
 """
 
+import asyncio
+import contextlib
 import json
 import re
 import sys
@@ -18,7 +27,7 @@ from ..client.chat import chat_keys, history, open_row, post
 from ..client.groups import my_groups, open_group
 from ..client.paths import group_arg, handle, resolve_group
 from ..client.roster import RosterCheck
-from ..errors import USAGE, CliError
+from ..errors import NODE, USAGE, CliError
 from ..output import date, err, safe, sentence
 from .context import Context
 
@@ -26,6 +35,10 @@ HELP = ("chat: alone, interactively; with a group, its last 10 messages (-N for
         "or post with -m")
 
 _COUNT = re.compile(r"^-(\d+)$")
+# --follow: how often the connection is asked whether it is alive, how long it
+# has to answer, and the waits between attempts to open it again.
+PING_EVERY, PING_TIMEOUT = 30, 10
+RECONNECT_WAITS = (2, 5, 10, 30)
 # The one verdict worth a line (roster.py): the application's words.
 KEY_CHANGED = ("! This account is using a key you have not seen before: a new device "
                "added elsewhere, or the node naming the wrong person.")
@@ -42,6 +55,9 @@ def add_parser(sub, parent) -> None:
                    help="how many of the last messages to show (also -N, as -100)")
     p.add_argument("-m", "--message", metavar="TEXT",
                    help="post TEXT instead of reading; - reads it from stdin")
+    p.add_argument("-f", "--follow", action="store_true",
+                   help="after the last messages, print new ones as they arrive, until "
+                        "Ctrl+C; with --json, one object per line")
 
 
 def rewrite_count(argv: list[str], actions) -> list[str]:
@@ -69,8 +85,10 @@ async def run(ctx: Context) -> int:
     a = ctx.args
     if a.last < 1:
         raise CliError(USAGE, "Ask for one message at least.")
+    if a.follow and a.message is not None:
+        raise CliError(USAGE, "Post with -m, or follow with -f; not both at once.")
     if a.group is None:
-        if a.message is not None:
+        if a.message is not None or a.follow:
             raise CliError(USAGE, "Name the group to post to.")
         if not (sys.stdin.isatty() and sys.stdout.isatty()):
             raise CliError(USAGE, "The interactive chat needs a terminal; name a group "
@@ -97,35 +115,168 @@ async def run(ctx: Context) -> int:
         check = RosterCheck(account, session)
         messages = [await open_row(r, keys, group["id"], check)
                     for r in await history(session, a.last)]
+        if a.follow:
+            follower = Follower(ctx, account, group, session, keys, check)
+            for m in messages:
+                follower.emit(m)
+            if not messages and not a.quiet:
+                err(f"No message in {safe(handle(group))} yet; waiting for one "
+                    "(Ctrl+C to stop).")
+            session = None                      # the follower owns it now
+            await follower.run()
+            return 0
     finally:
         if session is not None:
             await session.close()
         await account.close()
 
     if a.json:
-        json.dump([{"id": m.id, "timestamp": m.timestamp, "sender": m.sender,
-                    "text": m.text, "trust": m.trust,
-                    **({"unreadable": m.unreadable} if m.unreadable else {})}
-                   for m in messages],
-                  sys.stdout, indent=1, ensure_ascii=False)
+        json.dump([as_json(m) for m in messages], sys.stdout, indent=1, ensure_ascii=False)
         print()
         return 0
     if not messages and not a.quiet:
         err(f"No message in {safe(handle(group))} yet.")
     for m in messages:
-        head = f"{date(int(m.timestamp))}  {safe(m.sender)}: "
-        if m.unreadable:
-            print(f"{head}[this message cannot be read: {m.unreadable}]")
-            continue
-        lines = safe_lines(m.text)
-        print(head + lines[0])
-        for line in lines[1:]:
-            print(" " * len(head) + line)
-        if m.trust == "changed":
-            print(" " * len(head) + KEY_CHANGED)
+        print_message(m)
     return 0
 
 
+def print_message(m, *, flush: bool = False) -> None:
+    head = f"{date(int(m.timestamp))}  {safe(m.sender)}: "
+    if m.unreadable:
+        print(f"{head}[this message cannot be read: {m.unreadable}]", flush=flush)
+        return
+    lines = safe_lines(m.text)
+    print(head + lines[0])
+    for line in lines[1:]:
+        print(" " * len(head) + line)
+    if m.trust == "changed":
+        print(" " * len(head) + KEY_CHANGED)
+    if flush:
+        sys.stdout.flush()
+
+
+def as_json(m) -> dict:
+    return {"id": m.id, "timestamp": m.timestamp, "sender": m.sender, "text": m.text,
+            "trust": m.trust, **({"unreadable": m.unreadable} if m.unreadable else {})}
+
+
+class Follower:
+    """`--follow`: one connection at a time, opened again when it dies, and
+    every message printed once."""
+
+    def __init__(self, ctx: Context, account, group: dict, session, keys, check):
+        self.ctx, self.account, self.group = ctx, account, group
+        self.session, self.keys, self.check = session, keys, check
+        self.seen: set[str] = set()
+        self.last_ts = 0.0
+
+    def emit(self, m) -> None:
+        if m.key:
+            if m.key in self.seen:
+                return
+            self.seen.add(m.key)
+        self.last_ts = max(self.last_ts, m.timestamp)
+        if self.ctx.args.json:
+            print(json.dumps(as_json(m), ensure_ascii=False), flush=True)
+        else:
+            print_message(m, flush=True)
+
+    async def run(self) -> None:
+        try:
+            while True:
+                await self._listen()
+                await self.session.close()
+                self.session = None
+                if not self.ctx.args.quiet:
+                    err("meshbay-cli: the connection to the node was lost; reconnecting...")
+                await self._reopen()
+        finally:
+            if self.session is not None:
+                await self.session.close()
+
+    async def _listen(self) -> None:
+        """Print what the node pushes until the connection dies."""
+        s = self.session
+        queue: asyncio.Queue = asyncio.Queue()
+        dead = asyncio.Event()
+
+        def on_message(msg):
+            queue.put_nowait(msg)
+
+        def on_epoch(msg):
+            # Somebody was removed: the next message is sealed under a new key.
+            s.ack["chat_epoch"] = int(msg.get("epoch") or 0)
+            queue.put_nowait(None)
+
+        def on_state():
+            if s.pc is not None and s.pc.connectionState in ("failed", "closed"):
+                dead.set()
+
+        s.ch.on_push("chat_msg", on_message)
+        s.ch.on_push("chat_epoch_ack", on_epoch)
+        if s.pc is not None:
+            s.pc.on("connectionstatechange", on_state)
+        pinger = asyncio.ensure_future(self._ping(dead))
+        closed = asyncio.ensure_future(s.ch.closed.wait())
+        gone = asyncio.ensure_future(dead.wait())
+        try:
+            while True:
+                getter = asyncio.ensure_future(queue.get())
+                done, _ = await asyncio.wait({getter, closed, gone},
+                                             return_when=asyncio.FIRST_COMPLETED)
+                if getter not in done:
+                    getter.cancel()
+                    return
+                msg = getter.result()
+                if msg is None or int(msg.get("epoch") or 0) not in self.keys.by_epoch:
+                    with contextlib.suppress(CliError):
+                        self.keys = await chat_keys(s)
+                if msg is None:
+                    continue
+                msg = {**msg, "timestamp": msg.get("timestamp") or 0}
+                self.emit(await open_row(msg, self.keys, self.group["id"], self.check))
+        finally:
+            for task in (pinger, closed, gone):
+                task.cancel()
+            s.ch.off_push("chat_msg", on_message)
+            s.ch.off_push("chat_epoch_ack", on_epoch)
+
+    async def _ping(self, dead: asyncio.Event) -> None:
+        """A connection that stopped answering looks open to both ends."""
+        while True:
+            await asyncio.sleep(PING_EVERY)
+            try:
+                await self.session.ch.request({"type": "ping", "token": "follow"}, "pong",
+                                              timeout=PING_TIMEOUT)
+            except CliError:
+                dead.set()
+                return
+
+    async def _reopen(self) -> None:
+        """A new connection, then what was said while there was none."""
+        attempt = 0
+        while True:
+            try:
+                self.session = await open_group(self.account, self.group)
+                self.keys = await chat_keys(self.session)
+                self.check = RosterCheck(self.account, self.session)
+                rows = await history(self.session, 200)
+                break
+            except CliError as e:
+                if self.session is not None:
+                    await self.session.close()
+                    self.session = None
+                if e.code != NODE:
+                    raise
+                await asyncio.sleep(RECONNECT_WAITS[min(attempt, len(RECONNECT_WAITS) - 1)])
+                attempt += 1
+        for row in rows:
+            m = await open_row(row, self.keys, self.group["id"], self.check)
+            if m.timestamp >= self.last_ts:
+                self.emit(m)
+
+
 def safe_lines(text: str) -> list[str]:
     """A message's lines, each escaped: a line break stays a line break, every
     other control character is shown, never sent to the terminal."""
diff --git a/packages/meshbay-cli/src/meshbay_cli/manpage.py b/packages/meshbay-cli/src/meshbay_cli/manpage.py
index 022cd99..43a0fee 100644
--- a/packages/meshbay-cli/src/meshbay_cli/manpage.py
+++ b/packages/meshbay-cli/src/meshbay_cli/manpage.py
@@ -37,8 +37,8 @@ EXIT_CODES = {
                     "accepted by the node, a folder read-only or unavailable, leaving a "
                     "group you own, revoking this machine, deleting a file you may not "
                     "delete.",
-    errors.INTERRUPTED: "Interrupted (Ctrl+C). A download resumes where it stopped "
-                        "on the next run.",
+    errors.INTERRUPTED: "Interrupted (Ctrl+C), which is also how chat \\-\\-follow "
+                        "ends. A download resumes where it stopped on the next run.",
 }
 
 DESCRIPTION = r"""
@@ -232,6 +232,14 @@ before anything is asked or sent.
 shows the last messages of a group, oldest first, each with its time and the name
 its sender sealed in it, and posts one with
 .BR \-m .
+With
+.BR \-\-follow ,
+it then prints each new message as it arrives, until Ctrl+C (status 130); a
+connection that drops or stops answering is opened again, and what was said
+meanwhile is printed, each message once.
+With
+.BR \-\-json ,
+one object per line.
 Messages are encrypted and signed end to end, as the application's are; one this
 machine cannot verify or open is shown as unreadable, never left out.
 The device that signed a message is checked against the group's roster and the
@@ -263,7 +271,8 @@ On Windows,
 .IR %LOCALAPPDATA%\emeshbay\-cli\e .
 Holds
 .B state.json
-(the hub, the account name, pinned node keys, the browser access setting),
+(the hub, the account name, pinned node keys, the device keys accepted for
+the members whose messages were read, the browser access setting),
 .B secrets.enc
 (the session, the keys, invitation codes waiting for their node; encrypted),
 .B store.key
@@ -321,6 +330,7 @@ meshbay\-cli search \-t audio holiday
 meshbay\-cli search beach 2024 | head \-1 | xargs \-d '\\n' meshbay\-cli get
 meshbay\-cli rm 'Family/documents/draft*.odt'
 meshbay\-cli chat \-100 Family
+meshbay\-cli \-\-json chat Family \-\-follow | jq \-r .text
 echo 'On my way' | meshbay\-cli chat Family \-m \-
 meshbay\-cli devices approve Family@bob ZX4P\-7QRT
 meshbay\-cli leave \-\-yes Club@carol
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