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>

1 files changed, 168 insertions, 17 deletions

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."""