From 6856b682b1b9a6497da4fd145434bbb6b3dc2aae Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Mon, 6 Jul 2026 17:41:12 -0400 Subject: [PATCH] Add channel/direct chat feature and polish the passive-monitor UX MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce channel and direct messaging: an always-on ChatService that records inbound messages to history, a live full-screen chat screen, and a chat tool (interactive picker plus send/history/list/listen CLI). This bumps the DB schema to v3 and adds the supporting models, repository queries, and connection wiring. Also polish the monitor/menu UX: - Stop console log handlers from corrupting the full-screen TUI by detaching the RichHandler for the life of the session (file logging continues); most visible when toggling the monitor. - Colour only the header's monitor glyph (green ● on, red ○ off). - Crop the header to a single line instead of wrapping on narrow terminals. - Drop the redundant toggle-confirmation window so toggling returns straight to the main menu. Co-Authored-By: Claude Opus 4.8 --- README.md | 8 + meshtools/context.py | 21 +- meshtools/core/connection.py | 157 ++++++++ meshtools/core/models.py | 119 +++++++ meshtools/persistence/db.py | 21 +- meshtools/persistence/repository.py | 100 ++++++ meshtools/services/chat_service.py | 200 +++++++++++ meshtools/tools/chat.py | 460 ++++++++++++++++++++++++ meshtools/tools/monitor.py | 8 +- meshtools/ui/chat.py | 534 ++++++++++++++++++++++++++++ meshtools/ui/menu.py | 63 +++- meshtools/ui/tui/frame.py | 7 +- meshtools/ui/tui/render.py | 15 +- tests/test_chat.py | 487 +++++++++++++++++++++++++ 14 files changed, 2181 insertions(+), 19 deletions(-) create mode 100644 meshtools/services/chat_service.py create mode 100644 meshtools/tools/chat.py create mode 100644 meshtools/ui/chat.py create mode 100644 tests/test_chat.py diff --git a/README.md b/README.md index 7381f80..db811f7 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,7 @@ turned into interactive HTML visualizations. | **History / DB** | ✅ skeleton | Every tool execution and measurement is persisted and queryable. | | **TX-power optimization** | ✅ working | Sweep transmit power (coarse + refine), converge on the best signal, optionally apply, render an interactive chart. | | **Link budget** | ✅ working | Predict time-on-air, receiver sensitivity, range, and duty-cycle/dwell headroom for a radio config — offline, no radio needed. | +| **Chat** | ✅ working | Live full-screen channel and direct messaging — a scrolling transcript with a pinned input line where sent and received messages stream together. Every message is logged to SQLite, unread counts show in the menu header, and `chat send/history/list` script the same from the CLI. Channels are picked here; edit them in `config`. | | **Passive monitor** | ✅ working | Toggle a non-blocking background logger that records every overheard advert/telemetry (SNR, RSSI, location) to a longitudinal history while you keep using the app; on/off is remembered between sessions and live packet counts show in the menu. | | **Link health** | ✅ working | Compare a target's recent traces to its rolling baseline and flag SNR/reliability regressions, end-to-end and per hop. | | **Route map** | ✅ working | Aggregate trace history into an interactive route-stability graph — which links are stable, which the mesh flaps between, with churn metrics. | @@ -50,6 +51,13 @@ meshtools tx-optimize --target Alice --samples 6 --step 3 --apply meshtools info meshtools history +# Messaging: live chat in the menu, or scripted from the CLI +meshtools chat # interactive: pick a conversation, chat live +meshtools chat send --to Alice "on my way" # direct message +meshtools chat send --channel 0 "net in 5" # channel broadcast +meshtools chat history --to Alice +meshtools chat list + # Device configuration: view, set, back up, restore meshtools config # show all current settings meshtools config set radio_sf 9 # change one setting diff --git a/meshtools/context.py b/meshtools/context.py index 8753825..7894e15 100644 --- a/meshtools/context.py +++ b/meshtools/context.py @@ -22,6 +22,7 @@ from .persistence.logging import get_logger from .persistence.repository import Repository if TYPE_CHECKING: + from .services.chat_service import ChatService from .services.event_hub import EventHub from .services.monitor_service import MonitorService from .ui.surface import Ui @@ -62,6 +63,7 @@ class AppContext: _device: Optional[Device] = field(default=None, init=False, repr=False) _events: "Optional[EventHub]" = field(default=None, init=False, repr=False) _monitor: "Optional[MonitorService]" = field(default=None, init=False, repr=False) + _chat: "Optional[ChatService]" = field(default=None, init=False, repr=False) _ui: "Optional[Ui]" = field(default=None, init=False, repr=False) @property @@ -119,6 +121,21 @@ class AppContext: ) return self._monitor + @property + def chat(self) -> "ChatService": + """Return the session's chat service, creating it on first use. + + The service records inbound messages to history (as a subscriber of the always-on + event hub) and owns the outbound send path and the per-conversation unread counts. + It is created idle here; the interactive session starts it once a device is + available (and the live chat screen starts it lazily otherwise). + """ + if self._chat is None: + from .services.chat_service import ChatService + + self._chat = ChatService(self) + return self._chat + @property def log(self): # type: ignore[no-untyped-def] """The application logger.""" @@ -163,9 +180,11 @@ class AppContext: return self._device async def aclose(self) -> None: - """Stop monitoring, stop the event hub, disconnect the device, and close the repo.""" + """Stop monitoring and chat, stop the event hub, disconnect, and close the repo.""" if self._monitor is not None: await self._monitor.aclose() + if self._chat is not None: + await self._chat.aclose() if self._events is not None: await self._events.aclose() if self._device is not None: diff --git a/meshtools/core/connection.py b/meshtools/core/connection.py index 5b18034..ec9d885 100644 --- a/meshtools/core/connection.py +++ b/meshtools/core/connection.py @@ -14,6 +14,7 @@ Two implementations are provided: from __future__ import annotations import asyncio +import logging import random from abc import ABC, abstractmethod from datetime import datetime, timezone @@ -22,6 +23,9 @@ from typing import Callable, Optional from .events import MeshEvent from .models import Ack, Contact, Hop, Message, Observation, TraceResult +#: Module logger; enable DEBUG on ``meshtools.core.connection`` to trace the message pump. +_log = logging.getLogger(__name__) + #: Callback invoked with each :class:`~meshtools.core.events.MeshEvent` the device emits #: (an overheard packet, an inbound message, an acknowledgement). EventCallback = Callable[[MeshEvent], None] @@ -34,6 +38,14 @@ Unsubscribe = Callable[[], None] #: while a passive-monitor subscription is open (seconds). _MOCK_MONITOR_INTERVAL_S = 0.05 +#: How often the real device's inbound-message pump sweeps for queued messages, as a +#: safety net for firmware that doesn't reliably push ``MESSAGES_WAITING`` (seconds). +_MESSAGE_POLL_INTERVAL_S = 3.0 + +#: Per-``get_msg`` timeout in the message pump, so a missing device reply can't wedge the +#: drain loop (seconds). +_MESSAGE_GET_TIMEOUT_S = 5.0 + TX_POWER_MIN = 1 TX_POWER_MAX = 22 @@ -184,6 +196,36 @@ class Device(ABC): A zero-argument callable that stops the stream and releases the subscription. """ + # -- messaging --------------------------------------------------------------- + + @abstractmethod + async def send_direct_message(self, contact: Contact, text: str) -> Optional[Ack]: + """Send a direct text message to a contact. + + Args: + contact: The recipient; its ``public_key`` addresses the message. + text: The message body. + + Returns: + The delivery :class:`Ack` if one arrived before the send timed out, else + ``None`` (the message was handed to the radio but not yet acknowledged). + + Raises: + DeviceCommandError: If the companion rejected the send outright. + """ + + @abstractmethod + async def send_channel_message(self, index: int, text: str) -> None: + """Broadcast a text message on a channel slot. + + Args: + index: Zero-based channel slot to transmit on. + text: The message body. + + Raises: + DeviceCommandError: If the companion rejected the send. + """ + # -- configuration: extra reads --------------------------------------------- @abstractmethod @@ -679,7 +721,12 @@ class MeshCoreDevice(Device): if etype is not None: subs.append(subscribe(etype, ack_handler)) + # Drive the inbound-message pull ourselves (see ``_message_pump``): MeshCore never + # pushes message bodies, so without this sending works but nothing is received. + stop_pump = self._message_pump(mc, subs, subscribe) + def unsubscribe() -> None: + stop_pump() for sub in subs: unsub = getattr(sub, "unsubscribe", None) if unsub is not None: @@ -690,6 +737,103 @@ class MeshCoreDevice(Device): return unsubscribe + def _message_pump(self, mc, subs: list, subscribe) -> Unsubscribe: # type: ignore[no-untyped-def] + """Continuously pull inbound messages from the companion (the RX pull model). + + MeshCore doesn't push message bodies unsolicited: the device raises a + ``MESSAGES_WAITING`` notification and the client must call ``get_msg()`` to retrieve + each queued message, which the library's reader then dispatches as + ``CONTACT_MSG_RECV`` / ``CHANNEL_MSG_RECV`` to the handler registered above (a + command's own temporary listener does not consume the event — every subscriber + still sees it). We drive that pull three ways so it is robust across firmware + builds: an immediate drain (delivers anything already queued), a drain on each + ``MESSAGES_WAITING`` push (low latency), and a slow timer (a safety net for builds + whose pushes are unreliable — the failure this fixes). Drains are serialized by a + lock so the overlapping triggers never issue concurrent ``get_msg`` commands. + + Args: + mc: The connected ``MeshCore`` client. + subs: The subscription list to append the ``MESSAGES_WAITING`` sub to (so it is + torn down with the others). + subscribe: The client's ``subscribe`` callable. + + Returns: + A zero-argument callable that stops the pump (its poll task and drains). + """ + from meshcore import EventType + + stop = asyncio.Event() + draining = asyncio.Lock() + + async def drain() -> None: + # Pull until the device reports the queue is empty; each retrieved message is + # delivered to our handler by the reader's dispatch, so there's nothing to do + # with the returned event but check whether to keep going. + async with draining: + while not stop.is_set(): + try: + event = await mc.commands.get_msg(timeout=_MESSAGE_GET_TIMEOUT_S) + except Exception as exc: # noqa: BLE001 - transient; the poll retries + _log.debug("message pump: get_msg failed: %s", exc) + return + etype = getattr(event, "type", None) + if event is None or etype in (EventType.NO_MORE_MSGS, EventType.ERROR): + return + + def schedule_drain(_event=None) -> None: # noqa: ANN001 - MESSAGES_WAITING callback + asyncio.ensure_future(drain()) + + async def poll_loop() -> None: + while not stop.is_set(): + try: + await asyncio.wait_for(stop.wait(), _MESSAGE_POLL_INTERVAL_S) + except asyncio.TimeoutError: + await drain() # interval elapsed; sweep for anything the push missed + + waiting = getattr(EventType, "MESSAGES_WAITING", None) + if waiting is not None: + subs.append(subscribe(waiting, schedule_drain)) + schedule_drain() # immediate initial drain of anything already queued + poll_task = asyncio.ensure_future(poll_loop()) + + def stop_pump() -> None: + stop.set() # ends the poll loop and any in-flight drain at the next check + poll_task.cancel() + + return stop_pump + + async def send_direct_message( # noqa: D102 - inherited docstring + self, contact: Contact, text: str + ) -> Optional[Ack]: + from meshcore import EventType + + mc = self._require() + pub = self._node_pubkey(contact) + result = await mc.commands.send_msg(pub, text) + if result is None or getattr(result, "is_error", lambda: False)(): + raise DeviceCommandError( + f"failed to send message to {contact.name!r}: " + f"{getattr(result, 'payload', {})}" + ) + # The companion acknowledges the send immediately with an ``expected_ack`` code and + # a suggested wait; the recipient's delivery ACK arrives later carrying that code. + payload = getattr(result, "payload", {}) or {} + expected = payload.get("expected_ack") + expected_hex = expected.hex() if isinstance(expected, (bytes, bytearray)) else expected + if not expected_hex: + return None + suggested = payload.get("suggested_timeout") + timeout = suggested / 1000 * 1.2 if suggested else 8.0 + ack = await mc.wait_for_event( + EventType.ACK, attribute_filters={"code": expected_hex}, timeout=timeout + ) + return ack_from_event(ack) if ack is not None else None + + async def send_channel_message( # noqa: D102 - inherited docstring + self, index: int, text: str + ) -> None: + self._ok(await self._require().commands.send_chan_msg(index, text)) + @staticmethod def _ok(event): # type: ignore[no-untyped-def] """Return ``event`` if it succeeded, else raise its error payload. @@ -895,6 +1039,19 @@ class MockDevice(Device): async def set_tx_power(self, value: int) -> None: # noqa: D102 - inherited docstring self._tx_power = value + async def send_direct_message( # noqa: D102 - inherited docstring + self, contact: Contact, text: str + ) -> Optional[Ack]: + await asyncio.sleep(0) + # The simulator "delivers" instantly and always acknowledges, so outbound direct + # messages show as acked without a radio. + return Ack(code="mock") + + async def send_channel_message( # noqa: D102 - inherited docstring + self, index: int, text: str + ) -> None: + await asyncio.sleep(0) + async def admin_login(self, node: Contact, password: str) -> bool: # noqa: D102 await asyncio.sleep(0) if password != self._admin_password: diff --git a/meshtools/core/models.py b/meshtools/core/models.py index 975a5d0..1040621 100644 --- a/meshtools/core/models.py +++ b/meshtools/core/models.py @@ -118,6 +118,125 @@ class Ack: raw: Optional[dict] = None +def conversation_key( + is_channel: bool, channel_idx: Optional[int], peer: Optional[str] +) -> str: + """Return a stable key identifying a chat conversation. + + A conversation is either a channel or a direct exchange with one contact; this key + is how history, unread counts, and the live screen all agree on which thread a message + belongs to. + + Args: + is_channel: Whether the conversation is a channel. + channel_idx: The channel slot (for channel conversations). + peer: The contact's key prefix (for direct conversations). + + Returns: + ``"chan:"`` for a channel or ``"dm:"`` (lowercased) for a direct + exchange. + """ + if is_channel: + return f"chan:{channel_idx}" + return f"dm:{(peer or '').lower()}" + + +@dataclass(slots=True) +class ChatMessage: + """A chat message, sent or received, on a channel or with a contact. + + This is the persisted, display-oriented view of a message: it unifies the outbound + messages we send with the inbound :class:`Message` events the hub delivers, so a + conversation transcript is a single ordered list of these. Each belongs to exactly one + conversation — a channel (:attr:`is_channel` with :attr:`channel_idx`) or a direct + exchange with a contact (:attr:`peer` holding the contact's key prefix). + + Attributes: + text: The message body. + outbound: ``True`` if we sent it, ``False`` if we received it. + is_channel: Whether it belongs to a channel rather than a direct exchange. + channel_idx: The channel slot, for channel messages. + peer: The other party's key prefix, for direct messages. + peer_name: A friendly name for the peer/channel, snapshotted for display. + snr: Signal-to-noise ratio (dB) of an inbound reception, if reported. + acked: For an outbound direct message, whether delivery was acknowledged; ``None`` + when not applicable (a channel broadcast or an inbound message). + created_at: When the message was sent or received. + """ + + text: str + outbound: bool = False + is_channel: bool = False + channel_idx: Optional[int] = None + peer: Optional[str] = None + peer_name: Optional[str] = None + snr: Optional[float] = None + acked: Optional[bool] = None + created_at: datetime = field(default_factory=utcnow) + + @property + def key(self) -> str: + """The key of the conversation this message belongs to.""" + return conversation_key(self.is_channel, self.channel_idx, self.peer) + + @classmethod + def from_message( + cls, message: "Message", *, peer_name: Optional[str] = None + ) -> "ChatMessage": + """Build an inbound :class:`ChatMessage` from a received :class:`Message`. + + Args: + message: The inbound message delivered by the event hub. + peer_name: A friendly name for the sender, resolved from contacts if known. + + Returns: + The equivalent inbound :class:`ChatMessage`. + """ + return cls( + text=message.text, + outbound=False, + is_channel=message.is_channel, + channel_idx=message.channel, + peer=None if message.is_channel else (message.sender or None), + peer_name=peer_name, + snr=message.snr, + # Prefer the sender's own timestamp (the actual moment the message was + # composed) over ``received_at`` (when we happened to pull it off the radio), + # so the transcript reflects message time rather than retrieval time. Falls + # back to the receive time when the sender carried no timestamp. + created_at=message.sender_timestamp or message.received_at, + ) + + +@dataclass(slots=True) +class Conversation: + """A selectable chat thread: a channel or a direct exchange with a contact. + + Attributes: + label: Display name (e.g. ``#general`` or ``Alice``). + is_channel: Whether this is a channel rather than a direct conversation. + channel_idx: The channel slot, for channels. + contact: The contact, for direct conversations. + """ + + label: str + is_channel: bool + channel_idx: Optional[int] = None + contact: Optional[Contact] = None + + @property + def peer(self) -> Optional[str]: + """The peer key prefix for a direct conversation, else ``None``.""" + if self.is_channel or self.contact is None: + return None + return self.contact.key_prefix or self.contact.public_key[:12] or None + + @property + def key(self) -> str: + """The conversation's stable key (see :func:`conversation_key`).""" + return conversation_key(self.is_channel, self.channel_idx, self.peer) + + @dataclass(slots=True) class HeardNode: """Aggregated reception statistics for one node across many observations. diff --git a/meshtools/persistence/db.py b/meshtools/persistence/db.py index d8d0774..6154344 100644 --- a/meshtools/persistence/db.py +++ b/meshtools/persistence/db.py @@ -10,7 +10,7 @@ from __future__ import annotations import sqlite3 from pathlib import Path -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 _SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_meta ( @@ -90,11 +90,30 @@ CREATE TABLE IF NOT EXISTS observations ( observed_at TEXT NOT NULL ); +-- One chat message, sent or received, on a channel or with a contact. Unlike the other +-- measurement tables these are not tied to a single tool run (they arrive unsolicited via +-- the event hub), so run_id is nullable and detaches rather than cascades. +CREATE TABLE IF NOT EXISTS messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id INTEGER REFERENCES runs(id) ON DELETE SET NULL, + outbound INTEGER NOT NULL DEFAULT 0, + is_channel INTEGER NOT NULL DEFAULT 0, + channel_idx INTEGER, + peer TEXT, + peer_name TEXT, + text TEXT NOT NULL, + snr REAL, + acked INTEGER, + created_at TEXT NOT NULL +); + CREATE INDEX IF NOT EXISTS idx_traces_run ON traces(run_id); CREATE INDEX IF NOT EXISTS idx_trace_hops_trace ON trace_hops(trace_id); CREATE INDEX IF NOT EXISTS idx_tx_samples_run ON tx_samples(run_id); CREATE INDEX IF NOT EXISTS idx_observations_run ON observations(run_id); CREATE INDEX IF NOT EXISTS idx_observations_node ON observations(node); +CREATE INDEX IF NOT EXISTS idx_messages_channel ON messages(channel_idx); +CREATE INDEX IF NOT EXISTS idx_messages_peer ON messages(peer); """ diff --git a/meshtools/persistence/repository.py b/meshtools/persistence/repository.py index 6d2a7b0..1e8c7de 100644 --- a/meshtools/persistence/repository.py +++ b/meshtools/persistence/repository.py @@ -14,6 +14,7 @@ from pathlib import Path from typing import Any, Optional from ..core.models import ( + ChatMessage, HeardNode, Hop, Observation, @@ -382,3 +383,102 @@ class Repository: nodes = [HeardNode.from_observations(node, obs) for node, obs in grouped.items()] return sorted(nodes, key=lambda n: n.last_seen, reverse=True) + + # -- chat messages ---------------------------------------------------------- + + def record_chat_message( + self, msg: ChatMessage, *, run_id: Optional[int] = None + ) -> int: + """Persist one chat message (sent or received). + + Args: + msg: The message to store. Its ``peer`` is normalized to lowercase so a + direct conversation queries back consistently. + run_id: The owning background ``chat`` run, if any (inbound messages log to + one; outbound sends may not). + + Returns: + The new message's primary key. + """ + peer = msg.peer.lower() if msg.peer else None + cur = self._conn.execute( + "INSERT INTO messages " + "(run_id, outbound, is_channel, channel_idx, peer, peer_name, text, snr, " + "acked, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + run_id, + int(msg.outbound), + int(msg.is_channel), + msg.channel_idx, + peer, + msg.peer_name, + msg.text, + msg.snr, + None if msg.acked is None else int(msg.acked), + msg.created_at.isoformat(), + ), + ) + self._conn.commit() + return int(cur.lastrowid) + + def recent_chat_messages( + self, + *, + is_channel: bool, + channel_idx: Optional[int] = None, + peer: Optional[str] = None, + limit: int = 200, + ) -> list[ChatMessage]: + """Return a conversation's most recent messages, oldest-first. + + Args: + is_channel: Whether to load a channel conversation. + channel_idx: The channel slot (channel conversations). + peer: The contact key prefix (direct conversations). + limit: Maximum number of messages to return. + + Returns: + The messages in chronological order (ready to render as a transcript). + """ + if is_channel: + where, params = "is_channel = 1 AND channel_idx = ?", [channel_idx] + else: + where, params = "is_channel = 0 AND peer = ?", [(peer or "").lower()] + rows = self._conn.execute( + f"SELECT * FROM messages WHERE {where} ORDER BY id DESC LIMIT ?", + [*params, limit], + ).fetchall() + return [self._row_to_chat(row) for row in reversed(rows)] + + def last_chat_messages(self) -> dict[str, ChatMessage]: + """Return the latest message per conversation, keyed by conversation key. + + Backs the conversation picker's preview snippets. One row per distinct + conversation, taken as the highest-id (most recent) message in each. + + Returns: + A mapping of :func:`~meshtools.core.models.conversation_key` to its latest + :class:`ChatMessage`. + """ + rows = self._conn.execute( + "SELECT * FROM messages WHERE id IN (" + " SELECT MAX(id) FROM messages GROUP BY " + " CASE WHEN is_channel = 1 THEN 'chan:' || channel_idx " + " ELSE 'dm:' || peer END)" + ).fetchall() + return {msg.key: msg for msg in (self._row_to_chat(r) for r in rows)} + + @staticmethod + def _row_to_chat(row: sqlite3.Row) -> ChatMessage: + """Rebuild a :class:`ChatMessage` from a ``messages`` row.""" + return ChatMessage( + text=row["text"], + outbound=bool(row["outbound"]), + is_channel=bool(row["is_channel"]), + channel_idx=row["channel_idx"], + peer=row["peer"], + peer_name=row["peer_name"], + snr=row["snr"], + acked=None if row["acked"] is None else bool(row["acked"]), + created_at=datetime.fromisoformat(row["created_at"]), + ) diff --git a/meshtools/services/chat_service.py b/meshtools/services/chat_service.py new file mode 100644 index 0000000..af85290 --- /dev/null +++ b/meshtools/services/chat_service.py @@ -0,0 +1,200 @@ +"""The chat service: message persistence, unread tracking, and the send path. + +Like the passive monitor, this service is not a listener in its own right — the always-on +:class:`~meshtools.services.event_hub.EventHub` (``ctx.events``) does the listening. This +service is one of its subscribers: the one that writes every inbound +:class:`~meshtools.core.models.Message` to history as a +:class:`~meshtools.core.models.ChatMessage`, so a conversation transcript survives across +sessions. It also owns the per-conversation *unread* counters shown in the menu header, and +the outbound send path (so both the CLI and the live chat screen record what they send the +same way). + +The service is session-scoped state on the :class:`~meshtools.context.AppContext` +(``ctx.chat``). Unlike the monitor it has no on/off preference: messages addressed to us are +communications, not overheard noise, so they are always recorded while a device is listening. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +from ..core.connection import Unsubscribe +from ..core.events import EventKind, MeshEvent +from ..core.models import ChatMessage, Contact, Message, utcnow + +if TYPE_CHECKING: + from ..context import AppContext + + +class ChatService: + """Records inbound messages to history and tracks unread counts per conversation. + + Interact through the async lifecycle methods (:meth:`start`, :meth:`stop`, + :meth:`aclose`), the send helpers (:meth:`send_direct`, :meth:`send_channel`), and the + unread accessors. The currently-open conversation is registered via :meth:`set_active` + so its inbound messages don't inflate the unread badge. + """ + + def __init__(self, ctx: "AppContext") -> None: + """Initialize the service bound to an application context. + + Args: + ctx: The shared application context (device, repository, event hub, logger). + """ + self._ctx = ctx + self._unsubscribe: Optional[Unsubscribe] = None + self._run_id: Optional[int] = None + self._unread: dict[str, int] = {} + self._active: Optional[str] = None + self._session_count = 0 + + @property + def active(self) -> bool: + """Whether the service is subscribed to the hub and recording messages.""" + return self._unsubscribe is not None + + @property + def session_count(self) -> int: + """Number of inbound messages recorded since this process started.""" + return self._session_count + + def unread(self, key: str) -> int: + """Return the unread count for one conversation key.""" + return self._unread.get(key, 0) + + def unread_total(self) -> int: + """Return the total unread count across all conversations.""" + return sum(self._unread.values()) + + def set_active(self, key: Optional[str]) -> None: + """Mark ``key`` as the open conversation (or ``None`` when none is open). + + The open conversation is immediately cleared of unread and won't accrue more while + it stays active, since the user is looking at it. + + Args: + key: The conversation key now open, or ``None`` on close. + """ + self._active = key + if key is not None: + self._unread.pop(key, None) + + async def start(self) -> None: + """Begin recording inbound messages to history. Idempotent. + + Ensures the always-on event hub is running (which opens the device connection and + may raise if no device can be selected), then subscribes to its message stream and + opens a ``chat`` run for the recorded messages to link to. + + Raises: + Exception: Propagates any device/hub error; the caller can surface it and the + service simply stays inactive. + """ + if self.active: + return + await self._ctx.events.start() + run_id = self._ctx.repo.start_run( + "chat", {"mode": "background"}, self._ctx.profile_name + ) + + def on_event(event: MeshEvent) -> None: + # Runs on the event loop as messages arrive; keep it cheap and defensive so a + # single bad write can never take down the subscription. + message = event.message + if message is not None: + self._record_inbound(run_id, message) + + self._unsubscribe = self._ctx.events.subscribe(on_event, EventKind.MESSAGE) + self._run_id = run_id + self._ctx.log.info("chat recording (run %s)", run_id) + + async def stop(self) -> None: + """Stop recording and close the run record. Idempotent. + + A no-op if not recording. The event hub keeps listening; only this service's + recording subscription is removed. + """ + if not self.active: + return + try: + assert self._unsubscribe is not None + self._unsubscribe() + finally: + self._unsubscribe = None + if self._run_id is not None: + self._ctx.repo.finish_run( + self._run_id, "ok", {"messages": self._session_count} + ) + self._run_id = None + + async def aclose(self) -> None: + """Stop recording at session end.""" + await self.stop() + + def _record_inbound(self, run_id: int, message: Message) -> None: + """Persist one inbound message and bump its conversation's unread count. + + Args: + run_id: The owning background ``chat`` run. + message: The received message to record. + """ + chat = ChatMessage.from_message(message) + self._session_count += 1 + if chat.key != self._active: + self._unread[chat.key] = self._unread.get(chat.key, 0) + 1 + try: + self._ctx.repo.record_chat_message(chat, run_id=run_id) + except Exception as exc: # noqa: BLE001 - never let logging break the subscription + self._ctx.log.debug("chat: failed to record message: %s", exc) + + async def send_direct(self, contact: Contact, text: str) -> ChatMessage: + """Send a direct message to a contact and record it in history. + + Args: + contact: The recipient. + text: The message body. + + Returns: + The recorded outbound :class:`ChatMessage` (its ``acked`` reflects whether a + delivery acknowledgement arrived). + """ + device = await self._ctx.device() + ack = await device.send_direct_message(contact, text) + chat = ChatMessage( + text=text, + outbound=True, + is_channel=False, + peer=contact.key_prefix or contact.public_key[:12] or None, + peer_name=contact.name, + acked=ack is not None, + created_at=utcnow(), + ) + self._ctx.repo.record_chat_message(chat, run_id=self._run_id) + return chat + + async def send_channel( + self, index: int, text: str, *, label: Optional[str] = None + ) -> ChatMessage: + """Broadcast a message on a channel and record it in history. + + Args: + index: The channel slot to transmit on. + text: The message body. + label: A display label for the channel (e.g. ``#general``), stored for the + transcript. + + Returns: + The recorded outbound :class:`ChatMessage`. + """ + device = await self._ctx.device() + await device.send_channel_message(index, text) + chat = ChatMessage( + text=text, + outbound=True, + is_channel=True, + channel_idx=index, + peer_name=label, + created_at=utcnow(), + ) + self._ctx.repo.record_chat_message(chat, run_id=self._run_id) + return chat diff --git a/meshtools/tools/chat.py b/meshtools/tools/chat.py new file mode 100644 index 0000000..95e3b8b --- /dev/null +++ b/meshtools/tools/chat.py @@ -0,0 +1,460 @@ +"""The ``chat`` tool: channel and direct messaging over the mesh. + +Interactively it opens a conversation picker and then a live, full-screen chat (see +:mod:`meshtools.ui.chat`) where sent and received messages stream together. On the CLI it +exposes ``send``, ``history``, and ``list`` subcommands for scripted use. Channels are only +*listed* here for picking; creating and editing channel slots lives in the ``config`` tool. + +Every inbound message is recorded to history by the always-on +:class:`~meshtools.services.chat_service.ChatService`, and outbound messages are recorded on +send, so ``history`` reflects the full transcript regardless of which path produced it. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Optional + +import typer +from rich.table import Table +from rich.text import Text + +from ..context import AppContext +from ..core.connection import Device +from ..core.events import EventKind, MeshEvent +from ..core.models import ChatMessage, Contact, Conversation +from ..ui.tui import Choice, Separator +from .base import Tool, ToolResult, register + +#: How many channel slots to probe when listing channels to chat on. +_MAX_CHANNELS = 8 + +#: How many recent messages ``chat history`` prints by default. +_HISTORY_LIMIT = 50 + + +@register +class ChatTool(Tool): + """Send and receive channel and direct messages, with a live interactive chat.""" + + name = "chat" + help = "Channel and direct messaging — live chat, with stored history." + category = "Messaging" + order = 10 + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """Show the conversation picker and return the chosen thread to open. + + Args: + ctx: Shared application context. + + Returns: + ``{"conversation": Conversation}`` for the chosen thread, or ``None`` if the + user backed out. + """ + conversation = await self._pick_conversation(ctx) + if conversation is None: + return None + return {"conversation": conversation} + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Open a live chat (menu) or perform a scripted messaging action (CLI). + + Args: + ctx: Shared application context. + params: Either ``conversation`` (menu) or a ``cli_action`` with its arguments. + + Returns: + A :class:`ToolResult` summarizing what happened. + """ + action = params.get("cli_action") + if action is not None: + return await self._run_cli(ctx, action, params) + + conversation: Optional[Conversation] = params.get("conversation") + if conversation is None: # pragma: no cover - prompt_params returns None to cancel + return ToolResult(summary={}) + + from ..ui.chat import open_chat + + # Loop picker ↔ chat so backing out of a conversation (Esc) steps back to the + # picker rather than all the way to the main menu; Esc from the picker ends the + # loop and returns to the menu. The just-closed conversation is pre-selected in + # the picker so the cursor lands where the user left. + opened = 0 + shown = 0 + while conversation is not None: + shown = await open_chat(ctx, conversation) + opened += 1 + conversation = await self._pick_conversation(ctx, default_key=conversation.key) + return ToolResult(summary={"conversations": opened, "messages": shown}) + + # -- interactive picker ----------------------------------------------------- + + async def _pick_conversation( + self, ctx: AppContext, *, default_key: Optional[str] = None + ) -> Optional[Conversation]: + """Let the user pick a channel or contact to chat with. + + Args: + ctx: Shared application context. + default_key: Conversation key to pre-highlight (e.g. the one just backed out + of), so the cursor lands there rather than at the top. + + Returns: + The chosen :class:`~meshtools.core.models.Conversation`, or ``None`` if + cancelled. + """ + device = await ctx.device() + channels = await _read_channels(device) + contacts = await device.get_contacts() + lasts = ctx.repo.last_chat_messages() + + items: list = [Separator("── Channels ──")] + for conversation in channels: + items.append(Choice(title=_title(ctx, conversation, lasts), value=conversation)) + + items.append(Separator("── Direct ──")) + if contacts: + for contact in contacts: + conversation = Conversation(label=contact.name, is_channel=False, contact=contact) + items.append(Choice(title=_title(ctx, conversation, lasts), value=conversation)) + else: + items.append(Separator(" (no contacts yet — receive an advert first)")) + + items.append(Separator(" ")) + items.append(Choice(title="Back", value="__back__")) + + default = next( + ( + it.value + for it in items + if isinstance(it, Choice) + and isinstance(it.value, Conversation) + and it.value.key == default_key + ), + None, + ) + choice = await ctx.ui.select( + "Chat — pick a conversation", items, default=default + ) + if choice in (None, "__back__"): + return None + return choice + + # -- CLI -------------------------------------------------------------------- + + async def _run_cli( + self, ctx: AppContext, action: str, params: dict[str, Any] + ) -> ToolResult: + """Dispatch a scripted CLI action. + + Args: + ctx: Shared application context. + action: One of ``send``, ``history``, ``list``. + params: The action's arguments. + + Returns: + A :class:`ToolResult` for the action. + """ + if action == "send": + return await self._cli_send(ctx, params) + if action == "history": + return await self._cli_history(ctx, params) + if action == "listen": + return await self._cli_listen(ctx, params) + return await self._cli_list(ctx) + + async def _cli_send(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Send a channel or direct message and report the outcome.""" + device = await ctx.device() + text = str(params["text"]) + channel = params.get("channel") + if channel is not None: + await ctx.chat.send_channel(int(channel), text, label=f"#{channel}") + ctx.ui.note(f"[ok]✓[/ok] sent to channel [brand]{channel}[/brand]") + return ToolResult(summary={"channel": channel, "sent": True}) + + contact = _resolve_contact(await device.get_contacts(), str(params["to"])) + message = await ctx.chat.send_direct(contact, text) + state = "[ok]delivered[/ok]" if message.acked else "[warn]no ack[/warn]" + ctx.ui.note(f"[ok]✓[/ok] sent to [brand]{contact.name}[/brand] — {state}") + return ToolResult(summary={"to": contact.name, "acked": bool(message.acked)}) + + async def _cli_history(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Print a conversation's stored transcript.""" + limit = int(params.get("limit") or _HISTORY_LIMIT) + channel = params.get("channel") + to = params.get("to") + if channel is not None: + messages = ctx.repo.recent_chat_messages( + is_channel=True, channel_idx=int(channel), limit=limit + ) + label = f"#{channel}" + else: + device = await ctx.device() + contact = _resolve_contact(await device.get_contacts(), str(to)) + messages = ctx.repo.recent_chat_messages( + is_channel=False, + peer=contact.key_prefix or contact.public_key[:12], + limit=limit, + ) + label = contact.name + + if not messages: + ctx.ui.note(f"[muted]no messages with {label} yet[/muted]") + else: + ctx.ui.show(_history_table(label, messages)) + return ToolResult(summary={"conversation": label, "messages": len(messages)}) + + async def _cli_listen(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Tail inbound messages live in the console (and record them to history). + + A plain-console receiver, outside the full-screen menu: it starts the always-on + listener and prints each message as it arrives. Doubles as a diagnostic — with the + pump's debug logging enabled (``--debug`` on the CLI) it shows whether messages are + being pulled from the companion at all. + + Args: + ctx: Shared application context. + params: ``seconds`` — how long to listen (``0``/``None`` = until interrupted). + + Returns: + A :class:`ToolResult` with how many messages were seen. + """ + seconds = params.get("seconds") or 0 + await ctx.device() + await ctx.chat.start() # begins recording inbound to history too + ctx.console.print( + "[muted]listening for messages" + f"{f' for {seconds}s' if seconds else ' — press Ctrl-C to stop'}…[/muted]" + ) + seen = 0 + + def on_message(event: MeshEvent) -> None: + nonlocal seen + message = event.message + if message is None: + return + seen += 1 + stamp = message.received_at.astimezone().strftime("%H:%M:%S") + if message.is_channel: + who = f"#{message.channel}" + else: + who = message.sender or "?" + snr = f" [muted]({message.snr:+.0f} dB)[/muted]" if message.snr is not None else "" + ctx.console.print(f"[muted]{stamp}[/muted] [accent]{who}[/accent]: {message.text}{snr}") + + unsubscribe = ctx.events.subscribe(on_message, EventKind.MESSAGE) + try: + if seconds: + await asyncio.sleep(seconds) + else: + await asyncio.Event().wait() # until Ctrl-C / cancellation + except (KeyboardInterrupt, asyncio.CancelledError): # pragma: no cover - interactive + pass + finally: + unsubscribe() + return ToolResult( + summary={"messages": seen}, + message=f"[muted]stopped — saw {seen} message(s)[/muted]", + ) + + async def _cli_list(self, ctx: AppContext) -> ToolResult: + """List channels, contacts, and their most recent message.""" + device = await ctx.device() + channels = await _read_channels(device) + contacts = await device.get_contacts() + lasts = ctx.repo.last_chat_messages() + + table = Table(title="Conversations", border_style="muted", expand=False) + table.add_column("Conversation") + table.add_column("Unread", justify="right") + table.add_column("Last message") + rows = [*channels] + [ + Conversation(label=c.name, is_channel=False, contact=c) for c in contacts + ] + for conversation in rows: + last = lasts.get(conversation.key) + snippet = "" + if last is not None: + who = "you: " if last.outbound else "" + snippet = f"{who}{last.text[:40]}" + unread = ctx.chat.unread(conversation.key) + table.add_row(conversation.label, str(unread) if unread else "·", snippet) + ctx.ui.show(table) + return ToolResult(summary={"conversations": len(rows)}) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``chat`` subcommand group. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + chat_app = typer.Typer(help=self.help, no_args_is_help=True, rich_markup_mode="rich") + + @chat_app.command("send", help="Send a message to a contact or channel.") + def _send_cmd( + text: str = typer.Argument(..., help="The message body."), + to: Optional[str] = typer.Option(None, "--to", help="Contact name or key prefix."), + channel: Optional[int] = typer.Option(None, "--channel", help="Channel slot index."), + ) -> None: + if (to is None) == (channel is None): + raise typer.BadParameter("Pass exactly one of --to / --channel.") + run_tool_command( + self, {"cli_action": "send", "to": to, "channel": channel, "text": text} + ) + + @chat_app.command("history", help="Show a conversation's stored history.") + def _history_cmd( + to: Optional[str] = typer.Option(None, "--to", help="Contact name or key prefix."), + channel: Optional[int] = typer.Option(None, "--channel", help="Channel slot index."), + limit: int = typer.Option(_HISTORY_LIMIT, "--limit", help="Max messages to show."), + ) -> None: + if (to is None) == (channel is None): + raise typer.BadParameter("Pass exactly one of --to / --channel.") + run_tool_command( + self, + {"cli_action": "history", "to": to, "channel": channel, "limit": limit}, + ) + + @chat_app.command("list", help="List channels, contacts, and recent messages.") + def _list_cmd() -> None: + run_tool_command(self, {"cli_action": "list"}) + + @chat_app.command("listen", help="Tail inbound messages live in the console.") + def _listen_cmd( + seconds: int = typer.Option( + 0, "--seconds", "-s", help="How long to listen (0 = until Ctrl-C)." + ), + debug: bool = typer.Option( + False, "--debug", help="Log the message-pull activity (diagnostic)." + ), + ) -> None: + if debug: + _enable_receive_debug() + run_tool_command(self, {"cli_action": "listen", "seconds": seconds}) + + app.add_typer(chat_app, name=self.name) + + +# -- helpers ------------------------------------------------------------------ + + +def _enable_receive_debug() -> None: + """Route the message-pull and library debug logs to the console for diagnosis. + + Raises the ``meshcore`` and connection-layer loggers to DEBUG and attaches a stderr + handler, so ``chat listen --debug`` shows whether inbound messages are being pulled + from the companion (the receive path) at all. + """ + import logging + import sys + + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(logging.Formatter("%(name)s %(levelname)s %(message)s")) + for name in ("meshcore", "meshtools.core.connection"): + log = logging.getLogger(name) + log.setLevel(logging.DEBUG) + log.addHandler(handler) + + +async def _read_channels(device: Device) -> list[Conversation]: + """Probe channel slots and return them as channel conversations. + + Channel 0 (the default public channel) is always offered even when the firmware reports + no configured slots, so there is always somewhere to chat. + + Args: + device: The connected device to probe. + + Returns: + One :class:`~meshtools.core.models.Conversation` per configured channel (at least + channel 0). + """ + conversations: list[Conversation] = [] + for idx in range(_MAX_CHANNELS): + try: + channel = await device.get_channel(idx) + except Exception: # noqa: BLE001 - firmware may not support channel reads + break + if channel: + name = str(channel.get("channel_name") or idx) + conversations.append( + Conversation(label=f"#{name}", is_channel=True, channel_idx=idx) + ) + if not any(c.channel_idx == 0 for c in conversations): + conversations.insert(0, Conversation(label="#public", is_channel=True, channel_idx=0)) + return conversations + + +def _resolve_contact(contacts: list[Contact], needle: str) -> Contact: + """Find a contact by exact name (case-insensitive) or key-prefix match. + + Args: + contacts: The known contacts. + needle: A contact name or public-key prefix. + + Returns: + The matching contact. + + Raises: + typer.BadParameter: If no contact matches ``needle``. + """ + folded = needle.casefold() + for contact in contacts: + pub = (contact.public_key or "").lower().removeprefix("0x") + if contact.name.casefold() == folded or pub.startswith(folded): + return contact + raise typer.BadParameter(f"no contact matches {needle!r}") + + +def _title(ctx: AppContext, conversation: Conversation, lasts: dict) -> str: + """Build a picker row title: label, an unread marker, and a last-message snippet. + + Choice titles render as plain text, so markup is avoided; the unread count shows as a + ``●`` badge and the most recent message as a short trailing preview. + + Args: + ctx: Shared application context (for the unread count). + conversation: The conversation the row represents. + lasts: Map of conversation key to its most recent message. + + Returns: + A single-line plain-text title. + """ + parts = [conversation.label] + unread = ctx.chat.unread(conversation.key) + if unread: + parts.append(f" ● {unread}") + last: Optional[ChatMessage] = lasts.get(conversation.key) + if last is not None: + who = "you: " if last.outbound else "" + parts.append(f" · {who}{last.text[:32]}") + return "".join(parts) + + +def _history_table(label: str, messages: list[ChatMessage]) -> Table: + """Render a conversation's stored messages as a table. + + Args: + label: The conversation's display name. + messages: The messages to show, oldest-first. + + Returns: + A Rich :class:`Table` of time, sender, and text. + """ + table = Table(title=f"History — {label}", border_style="muted", expand=False) + table.add_column("Time") + table.add_column("From") + table.add_column("Message") + for message in messages: + stamp = message.created_at.astimezone().strftime("%m-%d %H:%M") + if message.outbound: + who = Text("you", style="accent") + else: + who = Text(message.peer_name or message.peer or "?", style="brand") + table.add_row(stamp, who, message.text) + return table diff --git a/meshtools/tools/monitor.py b/meshtools/tools/monitor.py index 6b22b14..f256a0c 100644 --- a/meshtools/tools/monitor.py +++ b/meshtools/tools/monitor.py @@ -107,11 +107,9 @@ class MonitorTool(Tool): "error": str(exc)}, message=f"[warn]monitoring enabled but capture couldn't start:[/warn] {exc}", ) - state = "[ok]● ON[/ok]" if now_on else "[muted]○ OFF[/muted]" - return ToolResult( - summary={"enabled": now_on, "active": ctx.monitor.active}, - message=f"passive monitor is now {state}", - ) + # No confirmation message: the new state is already shown live in the header's + # monitor indicator, so a result window here would just be redundant noise. + return ToolResult(summary={"enabled": now_on, "active": ctx.monitor.active}) @staticmethod def _view(ctx: AppContext) -> ToolResult: diff --git a/meshtools/ui/chat.py b/meshtools/ui/chat.py new file mode 100644 index 0000000..5c04bb7 --- /dev/null +++ b/meshtools/ui/chat.py @@ -0,0 +1,534 @@ +"""The live chat screen and its launcher. + +This is the interactive, full-screen chat experience: a scrolling transcript with an input +line pinned at the bottom, where messages you send and messages that arrive over the mesh +appear together in real time. Like the device picker and config editor, this module sits in +the UI layer but is allowed to depend on the context and services — it wires the +:class:`~meshtools.services.chat_service.ChatService` send path and the always-on event hub +to a :class:`~meshtools.ui.tui.screen.Screen`. + +The screen subscribes to the hub for the duration it is open so inbound messages for the +current conversation append live; :class:`~meshtools.services.chat_service.ChatService` +independently persists every inbound message, so history is complete whether or not the +screen is open. +""" + +from __future__ import annotations + +import asyncio +import re +from typing import TYPE_CHECKING, Awaitable, Callable, Optional + +from rich.console import Group, RenderableType +from rich.text import Text + +from ..core.events import EventKind, MeshEvent +from ..core.models import ChatMessage, Contact, Conversation, Message +from .theme import snr_style +from .tui.prompt import _LineEditor +from .tui.render import render_lines +from .tui.screen import CANCEL, Screen + +if TYPE_CHECKING: + from ..context import AppContext + +#: Fixed page step for PageUp/PageDown while scrolling the transcript (the screen isn't +#: told the viewport height, so a constant keeps paging predictable). +_PAGE = 10 + +#: How many past messages to load into the transcript when a conversation opens. +_HISTORY_LIMIT = 200 + +#: Palette of distinct, dark-theme-friendly colors cycled through to give each channel +#: sender its own hue (red is reserved for errors, so it's excluded). ``you`` and unknown +#: senders are styled separately. +_SENDER_COLORS = ( + "bold #f472b6", # pink + "bold #60a5fa", # blue + "bold #34d399", # green + "bold #a78bfa", # violet + "bold #fb923c", # orange + "bold #22d3ee", # cyan + "bold #a3e635", # lime + "bold #e879f9", # fuchsia +) + +#: Matches the ``Name: message`` convention channel senders use to identify themselves +#: (the protocol carries no sender field). The name is 1–20 non-colon characters and must +#: be followed by ``": "`` — conservative enough to leave ``http://…`` and ``note:x`` alone. +_SENDER_PREFIX = re.compile(r"^([^\s:][^:]{0,19}):[ \t]+(.*)$", re.DOTALL) + + +def _split_channel_sender(text: str) -> tuple[Optional[str], str]: + """Split a channel message into ``(sender_name, body)`` when it carries a name prefix. + + Channel messages have no sender field on the wire, so senders identify themselves by + prefixing the text with ``Name: ``. Lifting that name out lets the transcript show it + as a colored header and keep the body clean. + + Args: + text: The raw channel message text. + + Returns: + ``(name, body)`` when a plausible ``Name: `` prefix is present, else ``(None, text)``. + """ + match = _SENDER_PREFIX.match(text) + if match is None: + return None, text + name, body = match.group(1).strip(), match.group(2) + if not name or name.isdigit() or body.startswith("//"): # reject URLs / timestamps + return None, text + return name, body + + +class ChatScreen(Screen): + """A live conversation: a scrolling transcript above a pinned input line. + + The transcript auto-sticks to the newest message (and snaps back to the bottom whenever + you type or send). Scrolling up with the arrows / PageUp detaches from the bottom to + read history; End re-attaches. Enter sends the current line; Esc leaves the chat. + """ + + floating = False + + def __init__( + self, + conversation: Conversation, + messages: list[ChatMessage], + *, + send: Callable[[str], Awaitable[Optional[ChatMessage]]], + names: dict[str, str], + session, # noqa: ANN001 - TuiSession, imported lazily to avoid a cycle + ) -> None: + """Build the chat screen. + + Args: + conversation: The thread being shown (its label titles the screen). + messages: The initial transcript (history), oldest-first. + send: Async callable that sends a line and returns the recorded outbound + message (or ``None`` if nothing was sent). + names: Map of contact key prefix to friendly name, for labeling inbound + direct messages. + session: The running :class:`~meshtools.ui.tui.session.TuiSession`, used to + request repaints when messages arrive or a send completes. + """ + super().__init__() + self.title = conversation.label + self._is_channel = conversation.is_channel + self._messages = list(messages) + self._send = send + self._names = names + self._session = session + self._editor = _LineEditor() + self._sending = False + self._status = "" + self._stick = True # keep the newest message in view until the user scrolls up + # Channel-only reply selection: index of the highlighted message (or None when the + # compose line is focused), plus the body line it rendered on so the frame keeps it + # in view. Direct chats keep the flat, free-scrolling view and never select. + self._selected: Optional[int] = None + self._selected_line: Optional[int] = None + + @property + def footer_hint(self) -> str: + """Key hint, reflecting whether a message is picked for reply (channels only).""" + if not self._is_channel: + return "Enter send · ↑↓/PgUp scroll · End latest · Esc back" + if self._selected is not None: + return "Enter reply (@mention) · ↑↓ pick · End/Esc cancel" + return "Enter send · ↑ pick a message to reply · Esc back" + + # --- live updates -------------------------------------------------------- + + def append(self, message: ChatMessage) -> None: + """Append an inbound message to the transcript and repaint.""" + self._messages.append(message) + # Don't yank the view to the tail while the user is picking a message to reply to. + if self._selected is None: + self._stick = True + self._session.invalidate() + + # --- rendering ----------------------------------------------------------- + + def render_body(self, width: int) -> list[str]: + """Render the transcript, a divider, the input line, and any status.""" + self._selected_line = None + if not self._messages: + lines = render_lines(Text("No messages yet — say hello!", style="muted"), width) + elif self._is_channel: + # Channels carry many senders, so group consecutive messages from one sender + # under a colored header (Discord/Slack style) rather than the flat one-line + # format used for the two-party direct view. Rendering per-message also lets us + # record where the picked reply target lands (see _selected_line / cursor_line). + if self._selected is not None: + self._selected = max(0, min(self._selected, len(self._messages) - 1)) + lines = self._render_channel(width) + else: + lines = render_lines(Group(*(self._format(m) for m in self._messages)), width) + + footer_parts: list[RenderableType] = [ + Text("─" * width, style="muted"), + self._editor.render(), + ] + if self._is_channel and self._selected is not None: + footer_parts.append(self._reply_banner()) + if self._status: + footer_parts.append(Text(self._status, style="muted")) + lines += render_lines(Group(*footer_parts), width) + + # Sticking to the bottom: hand the frame an over-large offset so it clamps the view + # to the final lines (input + latest messages). Scrolling up clears the stick. + if self._stick: + self.scroll = len(lines) + return lines + + def cursor_line(self) -> Optional[int]: + """Keep the picked reply target in view; otherwise free scroll (managed by stick).""" + return self._selected_line + + def _format(self, message: ChatMessage) -> Text: + """Render one transcript message as ``HH:MM who text`` with quality cues.""" + stamp = message.created_at.astimezone().strftime("%H:%M") + if message.outbound: + who, style = "you", "accent" + elif message.is_channel: + who, style = (message.peer_name or "·"), "brand" + else: + who, style = (self._name(message.peer) or message.peer or "?"), "brand" + + line = Text() + line.append(f"{stamp} ", style="muted") + line.append(f"{who:>10.10} ", style=style) + line.append(message.text) + if message.outbound and message.acked is False: + line.append(" ⚠ no ack", style="warn") + if message.snr is not None: + line.append(f" {message.snr:+.0f} dB", style=snr_style(message.snr)) + return line + + def _name(self, peer: Optional[str]) -> Optional[str]: + """Resolve a sender key prefix to a contact name (exact, then prefix match).""" + if not peer: + return None + needle = peer.lower() + if needle in self._names: + return self._names[needle] + for prefix, name in self._names.items(): + if prefix.startswith(needle) or needle.startswith(prefix): + return name + return None + + # --- channel (grouped) rendering ----------------------------------------- + + def _render_channel(self, width: int) -> list[str]: + """Render the channel transcript to ANSI lines, tracking the picked message's row. + + Consecutive messages from the same sender on the same day share one colored header, + with each message body indented below. A muted divider marks each new day, and a + blank line separates distinct sender groups. Rendering message-by-message (rather + than as one Group) lets us record the body line of the selected reply target in + :attr:`_selected_line` so the frame can scroll it into view. + """ + lines: list[str] = [] + prev_sender: Optional[str] = None + prev_day = None + for idx, message in enumerate(self._messages): + stamp = message.created_at.astimezone() + day = stamp.date() + sender, body = self._sender_and_body(message) + new_day = day != prev_day + if new_day: + if lines: + lines += render_lines(Text(""), width) + lines += render_lines( + Text(f"── {stamp:%a} {stamp:%b} {stamp.day} ──", style="muted"), width + ) + if new_day or sender != prev_sender: + if not new_day and lines: + lines += render_lines(Text(""), width) # gap between sender groups + lines += render_lines(self._group_header(sender), width) + selected = idx == self._selected + if selected: + self._selected_line = len(lines) + lines += render_lines(self._body_line(body, message, selected=selected), width) + prev_sender, prev_day = sender, day + return lines + + def _sender_and_body(self, message: ChatMessage) -> tuple[str, str]: + """Return the display sender and cleaned body for a channel message.""" + if message.outbound: + return "you", message.text + name, body = _split_channel_sender(message.text) + return (name or "·"), body + + def _group_header(self, sender: str) -> Text: + """Build the sender header that starts a group, colored per sender.""" + return Text(sender, style=self._sender_style(sender)) + + def _body_line(self, body: str, message: ChatMessage, *, selected: bool = False) -> Text: + """Build one indented message line, each stamped with its own time. + + The per-message timestamp lives here (not on the group header) so every message + shows the time it was actually sent, even when several are grouped under one + sender — otherwise a run of same-sender messages would appear to share one time. + When ``selected``, the line is marked as the reply target (matching the select + screen's ``❯`` pointer and brand highlight). + """ + stamp = message.created_at.astimezone() + line = Text() + line.append("❯ " if selected else " ", style="brand" if selected else None) + line.append(f"{stamp:%H:%M} ", style="brand" if selected else "muted") + line.append(body, style="brand" if selected else None) + if message.outbound and message.acked is False: + line.append(" ⚠ no ack", style="warn") + if message.snr is not None: + line.append(f" {message.snr:+.0f} dB", style=snr_style(message.snr)) + return line + + def _reply_banner(self) -> Text: + """One-line cue shown above the input when a message is picked to reply to.""" + sender, _ = self._sender_and_body(self._messages[self._selected]) + who = sender if sender not in ("·", "you") else "this message" + return Text( + f"↩ Enter to reply to {who} with an @mention · End to cancel", style="accent" + ) + + def _sender_style(self, sender: str) -> str: + """Pick a stable color for a channel sender (accent for us, muted for unknown).""" + if sender == "you": + return "accent" + if sender == "·": + return "muted" + return _SENDER_COLORS[sum(map(ord, sender)) % len(_SENDER_COLORS)] + + # --- input --------------------------------------------------------------- + + def handle(self, action: str, data: str = "") -> None: + """Dispatch a key: channels navigate a reply selection; direct chats free-scroll.""" + if self._is_channel: + self._handle_channel(action, data) + else: + self._handle_flat(action, data) + + def _handle_flat(self, action: str, data: str = "") -> None: + """Direct-chat input: send on Enter, scroll the transcript, edit, or leave on Esc.""" + if action == "enter": + self._submit() + elif action == "escape": + self.resolve(CANCEL) + elif action == "up": + self._stick = False + self.scroll = max(0, self.scroll - 1) + elif action == "pageup": + self._stick = False + self.scroll = max(0, self.scroll - _PAGE) + elif action == "down": + self.scroll += 1 + elif action == "pagedown": + self.scroll += _PAGE + elif action == "home": + self._stick = False + self.scroll = 0 + elif action == "end": + self._stick = True + else: + if self._editor.edit(action, data): + self._stick = True # typing snaps back to the live tail + + def _handle_channel(self, action: str, data: str = "") -> None: + """Channel input: arrows pick a message to reply to; Enter sends or starts a reply. + + With no message picked the view behaves like the compose line (Enter sends). Moving + up enters the selection from the newest message; moving down past the newest (or + End) returns focus to the compose line and clears the selection. Enter on a picked + message primes the input with an ``@mention`` instead of sending. + """ + if action == "enter": + if self._selected is not None: + self._begin_reply() + else: + self._submit() + elif action == "escape": + if self._selected is not None: + self._clear_selection() # first Esc deselects; next leaves the chat + self._session.invalidate() + else: + self.resolve(CANCEL) + elif action == "up": + self._move_selection(-1) + elif action == "pageup": + self._move_selection(-_PAGE) + elif action == "down": + self._move_selection(1) + elif action == "pagedown": + self._move_selection(_PAGE) + elif action == "home": + if self._messages: + self._selected = 0 + self._stick = False + elif action == "end": + self._clear_selection() + self._stick = True # jump back to the live tail / compose line + else: + if self._editor.edit(action, data): + # Touching the compose line returns focus there — nothing stays selected. + self._clear_selection() + self._stick = True + + def _move_selection(self, delta: int) -> None: + """Move the reply selection by ``delta`` messages (negative = toward older). + + Entering from the compose line only happens moving up (``delta < 0``); moving past + the newest message drops the selection and re-sticks to the tail. + """ + if not self._messages: + return + last = len(self._messages) - 1 + if self._selected is None: + if delta < 0: + self._selected = last + self._stick = False + return + target = self._selected + delta + if target > last: + self._clear_selection() + self._stick = True + else: + self._selected = max(0, target) + self._stick = False + + def _clear_selection(self) -> None: + """Drop any reply selection (focus returns to the compose line).""" + self._selected = None + self._selected_line = None + + def _begin_reply(self) -> None: + """Prime the compose line with an ``@mention`` of the picked message's sender.""" + message = self._messages[self._selected] + sender, _ = self._sender_and_body(message) + mention = f"@[{sender}] " if sender and sender not in ("you", "·") else "@" + self._editor = _LineEditor(mention + self._editor.text) + self._clear_selection() + self._stick = True + self._session.invalidate() + + def _submit(self) -> None: + """Send the current input line as a message (scheduled off the key handler).""" + text = self._editor.text.strip() + if not text or self._sending: + return + self._editor = _LineEditor() + self._sending = True + self._status = "sending…" + self._stick = True + + async def _run() -> None: + try: + message = await self._send(text) + if message is not None: + self._messages.append(message) + self._status = "" + except Exception as exc: # noqa: BLE001 - report inline, keep the chat alive + self._status = f"send failed: {exc}" + finally: + self._sending = False + self._stick = True + self._session.invalidate() + + asyncio.ensure_future(_run()) + + +async def open_chat(ctx: "AppContext", conversation: Conversation) -> int: + """Open the live chat screen for ``conversation`` and run it until dismissed. + + Ensures listening and message recording are active, loads the stored transcript, + subscribes to the hub so inbound messages for this thread append live, and pushes the + screen. The subscription and the active-conversation marker are always cleaned up on + exit. + + Args: + ctx: The shared application context (must be running the interactive TUI surface). + conversation: The channel or contact conversation to open. + + Returns: + The number of messages shown in the transcript when the screen closed. + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + from .surface import TuiUi + + if not isinstance(ctx.ui, TuiUi): # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("live chat is only available in the interactive menu") + session = ctx.ui.session + + device = await ctx.device() + try: + await ctx.chat.start() # begin recording inbound if it wasn't already + except Exception: # noqa: BLE001 - the hub may already be running; recording is best-effort + pass + + history = ctx.repo.recent_chat_messages( + is_channel=conversation.is_channel, + channel_idx=conversation.channel_idx, + peer=conversation.peer, + limit=_HISTORY_LIMIT, + ) + names = _contact_names(await device.get_contacts()) + + async def send(text: str) -> Optional[ChatMessage]: + if conversation.is_channel: + assert conversation.channel_idx is not None + return await ctx.chat.send_channel( + conversation.channel_idx, text, label=conversation.label + ) + assert conversation.contact is not None + return await ctx.chat.send_direct(conversation.contact, text) + + screen = ChatScreen(conversation, history, send=send, names=names, session=session) + ctx.chat.set_active(conversation.key) + + def on_event(event: MeshEvent) -> None: + message = event.message + if message is not None and _belongs(message, conversation): + peer_name = names.get((message.sender or "").lower()) + screen.append(ChatMessage.from_message(message, peer_name=peer_name)) + + unsubscribe = ctx.events.subscribe(on_event, EventKind.MESSAGE) + try: + await session.run_screen(screen) + finally: + unsubscribe() + ctx.chat.set_active(None) + return len(screen._messages) + + +def _contact_names(contacts: list[Contact]) -> dict[str, str]: + """Build a key-prefix → name map for labeling inbound direct messages.""" + names: dict[str, str] = {} + for contact in contacts: + for key in (contact.key_prefix, contact.public_key[:12]): + if key: + names[key.lower()] = contact.name + return names + + +def _belongs(message: Message, conversation: Conversation) -> bool: + """Whether an inbound message belongs to the open conversation. + + Args: + message: The received message. + conversation: The conversation currently shown. + + Returns: + ``True`` if the message should append to this transcript. + """ + if conversation.is_channel: + return message.is_channel and message.channel == conversation.channel_idx + if message.is_channel: + return False + sender = (message.sender or "").lower() + peer = (conversation.peer or "").lower() + if not sender or not peer: + return False + return sender.startswith(peer) or peer.startswith(sender) diff --git a/meshtools/ui/menu.py b/meshtools/ui/menu.py index 43ff2ef..323c57c 100644 --- a/meshtools/ui/menu.py +++ b/meshtools/ui/menu.py @@ -9,16 +9,43 @@ prompts layer as dialogs, and its output appears in a bounded, scrollable result from __future__ import annotations +import logging +from contextlib import contextmanager +from typing import Iterator + +from rich.logging import RichHandler from rich.text import Text from .. import __version__ from ..context import AppContext +from ..persistence.logging import get_logger from ..tools import all_tools from .surface import TuiUi from .theme import make_console from .tui import Choice, Separator, TuiSession +@contextmanager +def _silence_console_logging() -> Iterator[None]: + """Detach console log handlers for the life of the full-screen TUI. + + The full-screen session owns the terminal via prompt_toolkit; any handler that + writes log lines to the same console (the :class:`RichHandler` installed by + :func:`configure_logging`) corrupts the frame — most visibly when toggling the + monitor, which connects the device and emits INFO records on demand. File logging is + untouched, so the JSON-lines record stays complete; the handlers are restored on exit. + """ + logger = get_logger() + detached = [h for h in logger.handlers if isinstance(h, RichHandler)] + for handler in detached: + logger.removeHandler(handler) + try: + yield + finally: + for handler in detached: + logger.addHandler(handler) + + def _header(ctx: AppContext) -> Text: """Build the persistent one-line header (target + live monitor status). @@ -37,9 +64,20 @@ def _header(ctx: AppContext) -> Text: target = ctx.selected_device.label else: target = "[muted]no device[/muted]" + unread = ctx.chat.unread_total() + chat_segment = f" · [accent]✉ {unread} unread[/accent]" if unread else "" + # Colour only the leading status glyph (● / ○) — green when the monitor is on + # (enabled), red when off — leaving the rest of the text muted as before. + status = ctx.monitor.status_text() + glyph, rest = status[:1], status[1:] + glyph_style = "ok" if ctx.monitor.enabled else "err" + # The frame crops this to a single line (see frame.compose_base), so a narrow terminal + # shows what fits and chops the rest rather than wrapping onto a second row. return Text.from_markup( f"[brand]MeshTools[/brand] [muted]v{__version__}[/muted] · " - f"[muted]device:[/muted] {target} · [muted]{ctx.monitor.status_text()}[/muted]" + f"[muted]device:[/muted] {target} · " + f"[{glyph_style}]{glyph}[/{glyph_style}][muted]{rest}[/muted]" + f"{chat_segment}" ) @@ -61,12 +99,14 @@ async def run_menu(ctx: AppContext) -> None: await _startup(ctx) await _menu_loop(ctx, session) finally: - # Stop history recording (closing its run record) and the always-on event - # hub, even on an unexpected exit. + # Stop history + chat recording (closing their run records) and the always-on + # event hub, even on an unexpected exit. await ctx.monitor.aclose() + await ctx.chat.aclose() await ctx.events.aclose() - await session.run(main()) + with _silence_console_logging(): + await session.run(main()) ctx.console.print("[muted]bye 73![/muted]") @@ -77,6 +117,7 @@ async def _menu_loop(ctx: AppContext, session: TuiSession) -> None: ctx: The shared application context. session: The running TUI session. """ + last_selection: str | None = None while True: items: list = [] current_category: str | None = None @@ -88,9 +129,14 @@ async def _menu_loop(ctx: AppContext, session: TuiSession) -> None: items.append(Separator(" ")) items.append(Choice(title="quit", value="__quit__")) - selection = await session.select("What would you like to do?", items) + # Re-highlight the tool the user just backed out of, so returning to the menu + # lands the cursor where they left rather than at the top. + selection = await session.select( + "What would you like to do?", items, default=last_selection + ) if selection in (None, "__quit__"): return + last_selection = selection await _run_selection(ctx, selection) @@ -132,6 +178,13 @@ async def _resume_monitor(ctx: AppContext) -> None: await ctx.events.start() except Exception: # noqa: BLE001 - surface via the header, don't crash the menu pass + # Record inbound messages from launch so the inbox and unread badge stay current + # even before the chat screen is opened. Deferred (like the hub) when connect on + # start is off; the chat screen starts it lazily then. + try: + await ctx.chat.start() + except Exception: # noqa: BLE001 - surface via the header, don't crash the menu + pass if not ctx.monitor.enabled: return try: diff --git a/meshtools/ui/tui/frame.py b/meshtools/ui/tui/frame.py index 60ce37d..08e1808 100644 --- a/meshtools/ui/tui/frame.py +++ b/meshtools/ui/tui/frame.py @@ -95,13 +95,14 @@ def compose_base( Returns: An ANSI string of exactly ``rows`` lines, each within ``cols`` columns. """ - header_lines = render_lines(header, cols) + # The header is a single status line: crop it to one row so a narrow terminal never + # wraps it onto a second line (which would push the panel down and misreport its height). + header_lines = render_lines(header, cols, no_wrap=True) header_h = len(header_lines) viewport = max(1, rows - header_h - 1 - 2) # minus footer(1) and panel border(2) panel = _panel(base, cols - 4, viewport, active=True) footer = Text.from_markup(f"[muted]{footer_hint}[/muted]") - group = Group(header, panel, footer) - lines = render_lines(group, cols) + lines = header_lines + render_lines(Group(panel, footer), cols) # Guarantee we never exceed the terminal height (pt would otherwise clip unpredictably). if len(lines) > rows: lines = lines[:rows] diff --git a/meshtools/ui/tui/render.py b/meshtools/ui/tui/render.py index c74cc50..7c71bd9 100644 --- a/meshtools/ui/tui/render.py +++ b/meshtools/ui/tui/render.py @@ -47,12 +47,14 @@ def _console(width: int) -> Console: return console -def render_to_ansi(renderable: RenderableType, width: int) -> str: +def render_to_ansi(renderable: RenderableType, width: int, *, no_wrap: bool = False) -> str: """Render a Rich renderable to an ANSI string at ``width`` columns. Args: renderable: Any Rich renderable (table, panel, text, group, markup string). width: Target width in columns; content wraps/pads to it. + no_wrap: When ``True``, keep the content on a single line — show what fits and + crop the overflow (with an ellipsis) instead of wrapping onto more rows. Returns: The rendered output as an ANSI-escaped string, without a trailing newline. @@ -60,11 +62,14 @@ def render_to_ansi(renderable: RenderableType, width: int) -> str: width = max(1, width) console = _console(width) with console.capture() as capture: - console.print(renderable, end="") + if no_wrap: + console.print(renderable, end="", no_wrap=True, overflow="ellipsis", crop=True) + else: + console.print(renderable, end="") return capture.get() -def render_lines(renderable: RenderableType, width: int) -> list[str]: +def render_lines(renderable: RenderableType, width: int, *, no_wrap: bool = False) -> list[str]: """Render a Rich renderable to a list of ANSI lines at ``width`` columns. Each returned line is an independently styled ANSI string, so the caller can slice a @@ -73,12 +78,14 @@ def render_lines(renderable: RenderableType, width: int) -> list[str]: Args: renderable: Any Rich renderable. width: Target width in columns. + no_wrap: When ``True``, keep the content to a single cropped line (see + :func:`render_to_ansi`). Returns: The rendered lines, newline-free. A trailing empty line (from the final newline) is dropped so line counts match visible rows. """ - ansi = render_to_ansi(renderable, width) + ansi = render_to_ansi(renderable, width, no_wrap=no_wrap) lines = ansi.split("\n") if lines and lines[-1] == "": lines.pop() diff --git a/tests/test_chat.py b/tests/test_chat.py new file mode 100644 index 0000000..d990c75 --- /dev/null +++ b/tests/test_chat.py @@ -0,0 +1,487 @@ +"""Tests for the chat feature: device send, persistence, and the chat service. + +All run against the :class:`MockDevice` simulator and a temporary database; no hardware. +""" + +from __future__ import annotations + +import asyncio +import logging +import re +from pathlib import Path + +import pytest + +#: Strip ANSI SGR escapes so rendered transcript lines can be asserted as plain text. +_ANSI = re.compile(r"\x1b\[[0-9;]*m") + + +def _strip_ansi(text: str) -> str: + """Remove ANSI color escapes from rendered lines for plain-text assertions.""" + return _ANSI.sub("", text) + +from meshtools.core.connection import MockDevice +from meshtools.core.events import MeshEvent +from meshtools.core.models import ( + ChatMessage, + Contact, + Conversation, + Message, + conversation_key, +) +from meshtools.persistence.repository import Repository +from meshtools.services.chat_service import ChatService +from meshtools.services.event_hub import EventHub +from meshtools.ui.chat import ChatScreen +from meshtools.ui.tui.screen import CANCEL + + +class _StubSession: + """A session stand-in that just counts repaint requests.""" + + def __init__(self) -> None: + self.invalidations = 0 + + def invalidate(self) -> None: + self.invalidations += 1 + + +class _StubContext: + """Minimal :class:`~meshtools.context.AppContext` stand-in for chat tests.""" + + def __init__(self, device: MockDevice, repo: Repository) -> None: + self._device = device + self.repo = repo + self.log = logging.getLogger("test.chat") + self.profile_name = None + self.events = EventHub(self) + + async def device(self) -> MockDevice: + await self._device.connect() + return self._device + + +@pytest.fixture() +def repo(tmp_path: Path) -> Repository: + r = Repository(tmp_path / "chat.db") + yield r + r.close() + + +# -- domain models ------------------------------------------------------------ + + +def test_conversation_key_distinguishes_channels_and_directs() -> None: + """Channel and direct keys are stable and case-insensitive on the peer.""" + assert conversation_key(True, 0, None) == "chan:0" + assert conversation_key(False, None, "AABBCC") == "dm:aabbcc" + + contact = Contact(name="Alice", public_key="d4e5" + "0" * 60, key_prefix="d4e5f6a7") + conv = Conversation(label="Alice", is_channel=False, contact=contact) + assert conv.key == "dm:d4e5f6a7" + assert conv.peer == "d4e5f6a7" + + chan = Conversation(label="#public", is_channel=True, channel_idx=2) + assert chan.key == "chan:2" + assert chan.peer is None + + +def test_chat_message_from_inbound_message() -> None: + """An inbound direct Message maps to a received ChatMessage on the right thread.""" + message = Message(text="hi", sender="a1b2c3d4", is_channel=False, snr=5.5) + chat = ChatMessage.from_message(message, peer_name="Yagi") + assert chat.outbound is False + assert chat.is_channel is False + assert chat.peer == "a1b2c3d4" + assert chat.peer_name == "Yagi" + assert chat.snr == 5.5 + assert chat.key == "dm:a1b2c3d4" + + +# -- device send -------------------------------------------------------------- + + +async def test_mock_send_direct_returns_ack() -> None: + """The simulator acknowledges direct messages so they show as delivered.""" + device = MockDevice() + await device.connect() + contact = (await device.get_contacts())[0] + ack = await device.send_direct_message(contact, "hello") + assert ack is not None + await device.send_channel_message(0, "hi channel") # no return, must not raise + await device.disconnect() + + +async def test_message_pump_drains_until_empty() -> None: + """The RX pump pulls get_msg() until the queue is empty (the pull model). + + MeshCore never pushes message bodies, so the client must pull them; without this, + sending works but nothing is received. Guards the immediate drain, the loop until + NO_MORE_MSGS, the MESSAGES_WAITING subscription, and clean teardown. + """ + from meshcore import EventType + + from meshtools.core.connection import MeshCoreDevice + + class _Ev: + def __init__(self, t) -> None: # noqa: ANN001 + self.type = t + + class _FakeCommands: + def __init__(self, script: list) -> None: + self.calls = 0 + self._script = script + + async def get_msg(self, timeout=None): # noqa: ANN001, ANN201 + i = min(self.calls, len(self._script) - 1) + self.calls += 1 + return _Ev(self._script[i]) + + class _FakeMC: + def __init__(self, script: list) -> None: + self.commands = _FakeCommands(script) + self.subs: list = [] + + def subscribe(self, etype, cb): # noqa: ANN001, ANN201 + self.subs.append(etype) + return object() + + # One real message, then the empty sentinel: the drain loop pulls twice. + mc = _FakeMC([EventType.CONTACT_MSG_RECV, EventType.NO_MORE_MSGS]) + device = MeshCoreDevice(port="COM_TEST") + subs: list = [] + stop = device._message_pump(mc, subs, mc.subscribe) + try: + await asyncio.sleep(0.01) # let the immediate drain task run + assert mc.commands.calls == 2 # CONTACT_MSG_RECV then NO_MORE_MSGS + assert EventType.MESSAGES_WAITING in mc.subs # low-latency push subscription + finally: + stop() + + +# -- persistence -------------------------------------------------------------- + + +def test_record_and_load_direct_conversation(repo: Repository) -> None: + """Direct messages round-trip and come back in chronological order.""" + repo.record_chat_message(ChatMessage(text="hi", outbound=True, peer="D4E5F6A7", acked=True)) + repo.record_chat_message(ChatMessage(text="yo", outbound=False, peer="d4e5f6a7", snr=3.0)) + + got = repo.recent_chat_messages(is_channel=False, peer="d4e5f6a7") + assert [m.text for m in got] == ["hi", "yo"] + assert got[0].outbound is True and got[0].acked is True + assert got[1].outbound is False and got[1].snr == 3.0 + + +def test_record_and_load_channel_conversation(repo: Repository) -> None: + """Channel messages are keyed by slot, isolated from direct messages.""" + repo.record_chat_message(ChatMessage(text="c0", is_channel=True, channel_idx=0)) + repo.record_chat_message(ChatMessage(text="c1", is_channel=True, channel_idx=1)) + + assert [m.text for m in repo.recent_chat_messages(is_channel=True, channel_idx=0)] == ["c0"] + assert [m.text for m in repo.recent_chat_messages(is_channel=True, channel_idx=1)] == ["c1"] + + +def test_last_chat_messages_returns_latest_per_conversation(repo: Repository) -> None: + """The picker preview shows the newest message in each conversation.""" + repo.record_chat_message(ChatMessage(text="old", peer="aa")) + repo.record_chat_message(ChatMessage(text="new", peer="aa")) + repo.record_chat_message(ChatMessage(text="chan", is_channel=True, channel_idx=0)) + + lasts = repo.last_chat_messages() + assert lasts["dm:aa"].text == "new" + assert lasts["chan:0"].text == "chan" + + +# -- chat service ------------------------------------------------------------- + + +async def test_service_records_inbound_and_tracks_unread(repo: Repository) -> None: + """Inbound messages are persisted and bump the conversation's unread count.""" + device = MockDevice() + ctx = _StubContext(device, repo) + chat = ChatService(ctx) + await chat.start() + try: + ctx.events.publish( + MeshEvent.message_event(Message(text="ping", sender="ffeeddcc", is_channel=False)) + ) + assert chat.unread("dm:ffeeddcc") == 1 + assert chat.unread_total() >= 1 + stored = repo.recent_chat_messages(is_channel=False, peer="ffeeddcc") + assert [m.text for m in stored] == ["ping"] + finally: + await chat.stop() + await device.disconnect() + + +async def test_service_active_conversation_suppresses_unread(repo: Repository) -> None: + """The open conversation clears and stops accruing unread while it stays active.""" + device = MockDevice() + ctx = _StubContext(device, repo) + chat = ChatService(ctx) + await chat.start() + try: + chat.set_active("dm:ffeeddcc") + ctx.events.publish( + MeshEvent.message_event(Message(text="ping", sender="ffeeddcc", is_channel=False)) + ) + assert chat.unread("dm:ffeeddcc") == 0 # active thread doesn't accrue unread + assert repo.recent_chat_messages(is_channel=False, peer="ffeeddcc") # still recorded + finally: + await chat.stop() + await device.disconnect() + + +# -- chat screen -------------------------------------------------------------- + + +def _screen(session: _StubSession, send, *, messages=None) -> ChatScreen: + """Build a ChatScreen for a direct conversation with a stub session and send hook.""" + conv = Conversation( + label="Alice", + is_channel=False, + contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"), + ) + return ChatScreen(conv, messages or [], send=send, names={"d4e5f6a7": "Alice"}, session=session) + + +def test_chat_screen_renders_transcript_and_input() -> None: + """The body shows each message and always ends with the input line.""" + session = _StubSession() + messages = [ChatMessage(text="hi there", outbound=True, peer="d4e5f6a7", acked=True)] + screen = _screen(session, send=None, messages=messages) + + lines = screen.render_body(60) + joined = "\n".join(lines) + assert "hi there" in joined + assert "›" in joined # the input editor's prompt marker + + +async def test_chat_screen_enter_sends_and_appends() -> None: + """Pressing Enter sends the line and appends the returned message to the transcript.""" + session = _StubSession() + sent: list[str] = [] + + async def send(text: str) -> ChatMessage: + sent.append(text) + return ChatMessage(text=text, outbound=True, peer="d4e5f6a7", acked=True) + + screen = _screen(session, send=send) + for ch in "hello": + screen.handle("text", ch) + screen.handle("enter") + await asyncio.sleep(0) # let the scheduled send task run + + assert sent == ["hello"] + assert screen._messages[-1].text == "hello" + assert session.invalidations > 0 + + +def test_chat_screen_scroll_detaches_and_end_reattaches() -> None: + """Scrolling up detaches from the live tail; End re-sticks to the bottom.""" + screen = _screen(_StubSession(), send=None) + assert screen._stick is True + screen.handle("up") + assert screen._stick is False + screen.handle("end") + assert screen._stick is True + + +def test_split_channel_sender_extracts_name_prefix() -> None: + """A ``Name: message`` channel line splits into sender and cleaned body.""" + from meshtools.ui.chat import _split_channel_sender + + assert _split_channel_sender("Alice: hey there") == ("Alice", "hey there") + assert _split_channel_sender("Yagi Repeater: online") == ("Yagi Repeater", "online") + # No plausible prefix: left untouched. + assert _split_channel_sender("just a message") == (None, "just a message") + assert _split_channel_sender("https://example.com") == (None, "https://example.com") + assert _split_channel_sender("14:30 standup") == (None, "14:30 standup") + + +def test_channel_transcript_groups_by_sender() -> None: + """Consecutive same-sender channel messages share one header; the body is cleaned.""" + from datetime import datetime, timezone + + conv = Conversation(label="#public", is_channel=True, channel_idx=0) + base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc) + + def at(minutes: int) -> datetime: + return base.replace(minute=24 + minutes) + + messages = [ + ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0, created_at=at(0)), + ChatMessage(text="Alice: again", is_channel=True, channel_idx=0, created_at=at(6)), + ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0, created_at=at(8)), + ChatMessage(text="hello all", outbound=True, is_channel=True, channel_idx=0, created_at=at(9)), + ] + screen = ChatScreen(conv, messages, send=None, names={}, session=_StubSession()) + rendered = _strip_ansi("\n".join(screen._render_channel(80))) + + assert rendered.count("Alice") == 1 # the two Alice messages share one header + assert "Bob" in rendered and "you" in rendered + assert "hi" in rendered and "again" in rendered # bodies present, prefix stripped + assert "Alice: hi" not in rendered # the raw name prefix is lifted into the header + # Each message keeps its own timestamp on its line, even when grouped under one sender. + stamps = [at(m).astimezone().strftime("%H:%M") for m in (0, 6, 8, 9)] + for stamp in stamps: + assert stamp in rendered + assert stamps[0] != stamps[1] # grouped Alice messages show distinct times + + +def _channel_screen(messages, session=None) -> ChatScreen: + """Build a channel ChatScreen over ``messages`` for selection/reply tests.""" + conv = Conversation(label="#public", is_channel=True, channel_idx=0) + return ChatScreen( + conv, messages, send=None, names={}, session=session or _StubSession() + ) + + +def _channel_messages(): + """Three inbound channel messages (Alice, Alice, Bob) for reply-selection tests.""" + from datetime import datetime, timezone + + base = datetime(2026, 7, 5, 14, 24, tzinfo=timezone.utc) + return [ + ChatMessage(text="Alice: hi", is_channel=True, channel_idx=0, created_at=base), + ChatMessage(text="Alice: still here", is_channel=True, channel_idx=0, created_at=base), + ChatMessage(text="Bob: yo", is_channel=True, channel_idx=0, created_at=base), + ] + + +def test_channel_up_enters_selection_from_newest() -> None: + """No message is selected until ↑ picks the newest, then steps toward older ones.""" + screen = _channel_screen(_channel_messages()) + assert screen._selected is None # compose focus: nothing selected initially + + screen.handle("up") + assert screen._selected == 2 # newest message + assert screen._stick is False + screen.handle("up") + assert screen._selected == 1 # steps to the previous message + + +def test_channel_down_past_newest_clears_selection() -> None: + """Moving ↓ past the newest message returns focus to compose (nothing selected).""" + screen = _channel_screen(_channel_messages()) + screen.handle("up") # select newest (index 2) + assert screen._selected == 2 + + screen.handle("down") # past the newest → deselect, re-stick to the tail + assert screen._selected is None + assert screen._stick is True + + +def test_channel_selected_line_tracked_for_scroll() -> None: + """Rendering records the selected message's body line so the frame keeps it in view.""" + screen = _channel_screen(_channel_messages()) + screen.handle("up") # select Bob (newest) + screen.render_body(80) + assert screen._selected_line is not None + # No selection → no cursor line, so the transcript free-scrolls as before. + screen.handle("end") + screen.render_body(80) + assert screen._selected_line is None + + +def test_channel_enter_on_selection_primes_at_mention() -> None: + """Enter on a picked message seeds the compose line with the sender's @mention.""" + screen = _channel_screen(_channel_messages()) + screen.handle("up") + screen.handle("up") # select an Alice message (index 1) + screen.handle("enter") + + assert screen._editor.text == "@[Alice] " + assert screen._selected is None # focus returns to compose after starting the reply + + +def test_channel_typing_clears_selection() -> None: + """Editing the compose line drops any reply selection (compose has focus).""" + screen = _channel_screen(_channel_messages()) + screen.handle("up") + assert screen._selected is not None + screen.handle("text", "x") + assert screen._selected is None + assert screen._editor.text == "x" + + +def test_chat_screen_escape_cancels() -> None: + """Esc resolves the screen's future with CANCEL so the caller pops it.""" + screen = _screen(_StubSession(), send=None) + loop = asyncio.new_event_loop() + try: + screen.future = loop.create_future() + screen.handle("escape") + assert screen.future.result() is CANCEL + finally: + loop.close() + + +async def test_service_send_records_outbound(repo: Repository) -> None: + """Sending through the service records the outbound message with its ack state.""" + device = MockDevice() + ctx = _StubContext(device, repo) + chat = ChatService(ctx) + await chat.start() + try: + contact = next(c for c in await device.get_contacts() if c.name == "Alice") + sent = await chat.send_direct(contact, "hey") + assert sent.outbound is True and sent.acked is True + + await chat.send_channel(0, "hello all", label="#public") + stored = repo.recent_chat_messages(is_channel=True, channel_idx=0) + assert [m.text for m in stored] == ["hello all"] + assert stored[0].outbound is True + finally: + await chat.stop() + await device.disconnect() + + +# -- end-to-end through the real session -------------------------------------- + + +async def test_open_chat_sends_through_real_session(tmp_path: Path) -> None: + """Driving open_chat with piped keys sends a message and records it, end to end.""" + from prompt_toolkit.input.defaults import create_pipe_input + from prompt_toolkit.output import DummyOutput + from rich.console import Console + + from meshtools.context import AppContext + from meshtools.core.admin_store import AdminStore + from meshtools.core.config import Settings + from meshtools.core.device_store import DeviceStore + from meshtools.ui.chat import open_chat + from meshtools.ui.surface import TuiUi + from meshtools.ui.tui.session import TuiSession + + settings = Settings(config_dir=tmp_path, db_path=tmp_path / "e2e.db") + ctx = AppContext( + console=Console(), + settings=settings, + repo=Repository(settings.db_path), + device_store=DeviceStore(tmp_path / "devices.json"), + admin_store=AdminStore(tmp_path / "admin.json"), + mock=True, + ) + conv = Conversation( + label="Alice", + is_channel=False, + contact=Contact(name="Alice", public_key="d4e5f6a7" + "0" * 56, key_prefix="d4e5f6a7"), + ) + try: + with create_pipe_input() as inp: + session = TuiSession(input=inp, output=DummyOutput()) + ctx.ui = TuiUi(session) + + async def main() -> None: + inp.send_text("hi\r\x1b") # type "hi", Enter (send), Esc (leave) + await open_chat(ctx, conv) + + await asyncio.wait_for(session.run(main()), timeout=5) + + stored = ctx.repo.recent_chat_messages(is_channel=False, peer="d4e5f6a7") + assert [m.text for m in stored] == ["hi"] + assert stored[0].outbound is True + finally: + await ctx.aclose()