Add channel/direct chat feature and polish the passive-monitor UX
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 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 4.8 committed
1 parent
20c32f9642
commit
6856b682b1
14 files changed
+2181
-19
No files matched your search
@@ -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
|
||||
|
||||
+20
-1
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:<idx>"`` for a channel or ``"dm:<peer>"`` (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.
|
||||
|
||||
@@ -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);
|
||||
"""
|
||||
|
||||
|
||||
|
||||
@@ -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"]),
|
||||
)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
+58
-5
@@ -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:
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user