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, 185 insertions, 20 deletions

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