# SPDX-License-Identifier: Apache-2.0 """The live chat screen and the function that opens it. This is the interactive, full-screen chat: a transcript that scrolls, with an input line pinned at the bottom. The messages that you send and the messages that come in over the mesh show together in real time. This module is in the UI layer, the same as the device picker and the config editor, but it can depend on the context and the services. It connects the send path of :class:`~meshterm.services.chat_service.ChatService` and the event hub, which always runs, to a :class:`~meshterm.ui.tui.screen.Screen`. While the screen is open, it subscribes to the hub. Thus the inbound messages for the current conversation are appended live. Separately, :class:`~meshterm.services.chat_service.ChatService` stores each inbound message. Thus the history is complete, also when the screen is not open. """ from __future__ import annotations import asyncio import re import unicodedata from collections.abc import Awaitable, Callable from functools import lru_cache from typing import TYPE_CHECKING, Any from rich.cells import cell_len from rich.console import Group, RenderableType from rich.text import Text from ..core import links, parts, reactions from ..core.channels import MENTION, split_channel_sender from ..core.connection import ContactNotOnDeviceError from ..core.events import EventKind, MeshEvent from ..core.models import ChatMessage, Contact, Conversation, Message, utcnow from ..core.regions import WILDCARD from ..platforms import Platform, on_platform from .compose import MentionEditor from .pathline import PATH_INK, PathHop from .qr import QrScreen, qr_strip from .theme import emoticon, name_style, snr_style from .tui.prompt import ( CHANNEL_BYTE_LIMIT, DM_BYTE_LIMIT, byte_counter, ) from .tui.render import render_hanging, render_lines, right_aligned_tail from .tui.screen import CANCEL, Screen from .tui.spinner import Spinner, spinner_interval from .widgets import ( DESTINATION, ORIGIN, author_chip, hop_chip, mention_hop, mention_mark, picked_hop, self_chip, ) if TYPE_CHECKING: from ..context import AppContext from .map_render import MapMarker from .map_screen import Locator from .node_detail_screen import ContactPages #: The delivery marks for a *resolved* outbound direct message, at the end of its line: #: acknowledged, or transmitted but not acknowledged (^R can retry it). These are the #: ``✓``/``✗`` status marks of the app in their ok/err styles, not the ✅/❌ emoji, which #: are for the lane of packet-class icons. While the ack is still pending, the line shows an #: animated spinner instead (refer to :meth:`ChatScreen._delivery_glyph`). _DELIVERED = ("✓", "ok") _FAILED = ("✗", "err") #: A word between whitespace. The wrap moves a word only as a whole. Thus the run that a #: wrap must not cut a URL in is the word that has the URL, with a bracket or a full stop #: next to it. _WORD = re.compile(r"\S+") #: The column where a URL starts when it is too long to hang under its body: the column #: after the ``❯`` that is in front of the first line of the selected message. A message #: that names us has its mark in this column, and the timestamp follows it. _URL_GUTTER = 2 #: Whether the URLs of a message are drawn as QR codes under it (``Platform.url_codes``, #: only on the PicoCalc). Bound for each platform, never asked at each frame. _URL_CODES = False #: The mark in front of the count of a reaction. Where the font has no emoji, an emoticon #: such as ``+1`` or ``<3`` stands in for the emoji, and a bare count after it reads as part #: of it (``+1 3``). There, the count is ``x3``. Bound for each platform. _COUNT_MARK = "" @on_platform def _bind(platform: Platform) -> None: """Bind the transcript to the platform (now and at each switch). The platform says whether codes hang under URLs, and how a reaction count is marked. """ global _URL_CODES, _COUNT_MARK _URL_CODES = platform.url_codes _COUNT_MARK = "" if platform.emoji else "x" #: The maximum width of a footer hint, in cells (refer to the UX standards). _HINT_CELLS = 72 def _fit_hint(atoms: list[str], spare: tuple[str, ...] = ()) -> str: """Join the hint atoms that are not empty, and remove ``spare`` atoms until the hint fits. The ``spare`` atoms go in the given order, and only while the hint is wider than :data:`_HINT_CELLS`. Thus a key that the line can hold stays on it. """ kept = [atom for atom in atoms if atom] for atom in spare: if cell_len(" · ".join(kept)) <= _HINT_CELLS: break if atom in kept: kept.remove(atom) return " · ".join(kept) #: The long words of a hint atom, and the short words that the same atom has when the hint #: needs the room. The two forms count as one atom. The short form leaves out a note or a #: second key, as the hint already does for ^U, ^L, and ^O. _SHORT_ATOMS = {"Enter reply (@mention)": "Enter reply", "^End/Esc cancel": "Esc cancel"} def _atom_set(hint: str) -> set[str]: """The atoms of a hint, each in its short form, to compare two hints by their atoms.""" return {_SHORT_ATOMS.get(atom, atom) for atom in hint.split(" · ")} class ChatScreen(Screen): """A live conversation: a transcript that scrolls, above a pinned input line. The transcript automatically stays on the newest message (and goes back to the bottom each time you type or send). ↑ selects a message. The selection moves with ↑↓/PgUp/PgDn, and the viewport moves with it. ^End (or Esc) puts the focus on the compose line again. Enter sends the current line, or does an action on a selected message: in a channel, it starts a reply with an ``@mention``, and in a direct chat, it opens the delivery paths of the message. In the two types of chat, ^P opens the paths of the selected message. A path belongs to one message, so when no message is selected, there is nothing to show. ^U shows the links of the selected message as QR codes, on the share screen. ^L opens the map on the node of the author of the selected message, when that node has a location. ^K locks that author against the archive sweep, or unlocks it. ^E reacts to the selected message with an emoji (refer to :mod:`~meshterm.core.reactions`). When no message is selected, Esc leaves the chat. A reaction is a message on the wire, but the transcript does not show it as one. It shows each reaction under the message that it reacts to: the emoji and a count, or the names of the nodes that reacted while that message is selected. A room has no reactions, so a room shows each message as it arrived. A channel with a send scope shows its region in the title (``#ops · yul``). There, ^R sends a message again *unscoped*: the selected message, or the newest when no message is selected. This is the solution for a scoped message that no repeater in range relays. It comes after an amber confirm, because an unscoped flood gets to each repeater that the scope kept it from. """ floating = False #: Home and End move the caret of the compose line here. Thus edge scroll does not use #: them. home_end_jumps = False @property def picocalc_lyra_lane(self): """The PicoCalc lane of a conversation: the same lane in a channel, a DM, and a room. The plain bank of F1 to F3 has the verbs that all three screens share, in the order of the hint: ``Paths`` (^P), ``URL→QR`` (^U), and ``Map`` (^L). The Shift bank of F1 and F2 has the verbs of one screen only. ``Retry`` (^R) is on the Shift half of F1 in a DM and in a room, and ``Resend`` (the same ^R, unscoped) is there in a channel. A room adds ``Log in`` (^G) on the Shift half of F2 (refer to :class:`~meshterm.ui.room.RoomScreen`). Thus a key does the same thing on each of the three screens, and the user finds a verb in the same place (JP, 2026-10-10). The shared verbs act *on* the selected message. ``Paths`` is lit while a message is selected. ``URL→QR`` shows the codes of the links of the selection on a share screen, one code at a time with its full URL, and is lit while the selection has a URL (refer to :meth:`_picked_urls`). On the Cardputer it is the only way to a code. On the PicoCalc, the transcript already draws each code under its message in the same size, so there the chip only shows one code alone. ``Map`` is lit while the author of the selection has a location (refer to :meth:`_locate_target`). ``Author`` (^O, the node page of the author of the selection) is the fourth shared verb, and the plain bank has only three slots. Thus it is on the Shift half of F3, behind ``Map``, the other verb about the author. It is lit while the author has a page (refer to :meth:`_author_page`). Each of the two halves of F3 is empty when the screen has nothing to find that node with. ``Lock`` (^K) is the fifth verb about the author, and it is on the Shift half of F2. It toggles the lock of the author, so it reads ``Unlock`` on a locked author, and it is lit while the author can have a lock (refer to :meth:`_lock_target`). That slot is free in a channel and in a DM. In a room, ``Log in`` has it, so there the user locks from the ``Author`` page, which a conversation opens with its lock row (JP, 2026-10-19). The slot is empty where MeshTerm cannot store a lock. ``React`` (^E) has no slot. It was on the Shift half of F2 before ``Lock`` came, and the Shift bank has no free slot now. ``Lock`` keeps the slot (Rich, 2026-10-22), so a reaction is ^E only, and the hint names it when there is room. The Shift half of F1 follows the two rules of the lane. In a channel with no resend connected, it is empty, because a channel message is never acknowledged and there is nothing to retry. In a DM with nothing outstanding, it is only dim. ``Resend`` sends the selected message (or the newest) again, unscoped (refer to :meth:`_retry_target`), and it is dim until that message is one of ours that went out under a region. The day jump (``ctrl_pageup``/``ctrl_pagedown``) has no chip. ``Day ↑`` and ``Day ↓`` were on F1 and F2, and JP removed them (2026-10-10). The desktop keeps the jump on ^PgUp and ^PgDn. F4/F5 keep the shared paging pair. The Shift bank of the shared lane sends the plain ``end``/``home`` actions. But here the cursor of the compose line already uses them (refer to :meth:`handle`). Thus, without a change, the two keyboard keys go to the same fallback ("clear the selection, stay at the tail"), and each one seems to scroll to the bottom. Instead, the two Shift slots send ``ctrl_end``/``ctrl_home``, and their labels say what these actions do to a conversation: go back to the *latest* message and the compose line, or go back to the *oldest*. Each jump is still behind the page key that goes in the same direction (refer to :data:`~meshterm.ui.tui.fkeys.DEFAULT_LANE`). From the outside, the transcript moves by one screen, also when it is the selection that moves. """ from .tui.fkeys import FPair, default_lane lane = list(default_lane(nav=bool(self._messages))) live = bool(self._messages) if not self._is_channel: again = ("Retry", "retry") elif self._resend_unscoped is not None: again = ("Resend", "retry") else: again = ("", "") lane[0] = FPair( "Paths", "paths", *again, enabled=self._selected is not None and self._paths is not None, opp_enabled=self._retry_target() is not None, ) lock = self._lock_target() if self._pages is not None and self._pages.can_lock: toggle = ("Unlock" if lock is not None and lock[1] else "Lock", "lock") else: toggle = ("", "") lane[1] = FPair( "URL→QR", "url_code", *toggle, enabled=bool(self._picked_urls()), opp_enabled=lock is not None, ) if self._locator is not None or self._pages is not None: where = ("Map", "locate") if self._locator is not None else ("", "") who = ("Author", "author_page") if self._pages is not None else ("", "") lane[2] = FPair( *where, *who, enabled=self._locate_target() is not None, opp_enabled=self._author_page() is not None, ) lane[3] = FPair("Page ↓", "pagedown", "Latest", "ctrl_end", enabled=live, opp_enabled=live) lane[4] = FPair("Page ↑", "pageup", "Oldest", "ctrl_home", enabled=live, opp_enabled=live) return lane def __init__( self, conversation: Conversation, messages: list[ChatMessage], *, send: Callable[[str], Awaitable[ChatMessage | None]], names: dict[str, str], session, # noqa: ANN001 - a TuiSession, imported late to prevent an import cycle resend: Callable[[ChatMessage], Awaitable[ChatMessage]] | None = None, paths: Callable[[ChatMessage], Awaitable[None]] | None = None, key_of: Callable[[str], str | None] | None = None, scope: str | None = None, resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None, locator: Locator | None = None, pages: ContactPages | None = None, self_name: str | None = None, reaction_set: list[str] | None = None, max_parts: int = 1, part_gap_s: float = 0.0, self_key: str | None = None, ) -> None: """Build the chat screen. Args: conversation: The thread that the screen shows (its label is the title of the screen). messages: The initial transcript (history), the oldest first. send: An async callable that sends a line and returns the stored outbound message (or ``None`` if nothing was sent). names: A map from contact key prefix to friendly name, for the labels of inbound direct messages. session: The running :class:`~meshterm.ui.tui.session.TuiSession`. The screen uses it to ask for a paint when messages arrive or a send completes. resend: An async callable that tries again to deliver a direct message that is not acknowledged, and changes the message in place (direct chats only, ``None`` for channels). paths: An async callable that shows the delivery paths of the selected message (the ^P view). ``None`` makes this action do nothing, with no message. key_of: Finds the key of the node of a sender from its display name (refer to :func:`~meshterm.services.trace_runner.make_name_key_resolver`). The key is the source of the hue of the sender. With ``None`` (or for a name that it cannot find), the sender gets the grey of an unknown node (``node.unknown``), because colour is only for identities that have a key. scope: The send scope of the channel, shown in the title as a bare ``·`` atom (``#ops · yul``: the title of a channel has no other atom that the user can think is a region name). ``None`` for a channel that sends under the default scope of the device, and always ``None`` for a direct chat. resend_unscoped: An async callable that sends the text of a channel message again, unscoped, and returns the stored message (channels only. With ``None``, ^R does nothing there). locator: The located nodes, for ^L: the map on the author of the selected message (refer to :class:`~meshterm.ui.map_screen.Locator`). With ``None``, ^L does nothing. pages: The contacts, for ^O: the node page of the author of the selected message (refer to :class:`~meshterm.ui.node_detail_screen.ContactPages`). With ``None``, ^O does nothing. self_name: The name of our node (refer to :func:`our_node`). Our messages and the mentions of our name show it (:func:`~meshterm.ui.widgets.self_chip`). A channel reaction names the author of its target, and our node is the author of each message that we sent. With ``None``, our channel messages get no reactions. reaction_set: The emoji that ^E offers, in order. ``None`` offers :data:`~meshterm.core.reactions.DEFAULT_REACTIONS`. max_parts: The largest number of parts for a message that is too long for one packet (refer to :mod:`meshterm.core.parts`). With 1, a message that is too long is refused. part_gap_s: The time to wait between two parts of a channel message, so that the parts do not go out as a burst. A direct part waits for its acknowledgement, and does not wait this time. self_key: Our public key, which gives the hue of that chip. """ super().__init__() self._self_name = self_name self._self_key = self_key self._scope = scope if conversation.is_channel else None self.title = f"{conversation.label} · {self._scope}" if self._scope else conversation.label self._is_channel = conversation.is_channel # Many voices or one. The user reads a channel and a room in the same way: each # message under its author, and Enter on a message starts a reply with an # @mention. A direct chat has two persons, and Enter on a message opens its paths. self._multiparty = conversation.is_channel or conversation.is_room # Whether typing goes to the compose line. It is off where no typed text can be # sent (a room that keeps nothing from us). Thus the keyboard keys never fill a # line that the user cannot see. self._composing = True self._key_of: Callable[[str], str | None] = key_of or (lambda name: None) # The one remote sender of a direct thread is the peer. Its key gives the colour of # the header, also when the resolver cannot find the display name. contact = conversation.contact self._peer_key = ( "" if conversation.is_channel or contact is None else (contact.public_key or contact.key_prefix or "") ) # The reactions of the conversation, folded onto their targets. A room has none. self._fold = ( None if conversation.is_room else reactions.ReactionFold(channel=conversation.is_channel, self_name=self_name) ) self._reaction_set = reaction_set or reactions.DEFAULT_REACTIONS.split() self._peer_label = conversation.label self._messages = [m for m in messages if not self._folds(m)] self._send = send self._resend = resend self._resend_unscoped = resend_unscoped if conversation.is_channel else None self._max_parts = max(1, max_parts) self._part_gap_s = max(0.0, part_gap_s) self._paths = paths self._locator = locator self._pages = pages self._names = names self._session = session self._editor = MentionEditor() self._sending = False # It turns while a direct message is in flight, so that the glyph at the end of the # message spins (instead of a static hourglass) until the ack resolves. All messages # share it: only one send or retry is ever in flight at a time (``_sending`` # controls the two). self._spinner = Spinner() self._status = "" self._stick = True # keep the newest message visible until the user scrolls up self._paths_open = False # one paths dialog at a time self._map_open = False # one map at a time self._page_open = False # one node page at a time self._codes_open = False # one share screen of links at a time self._paste_open = False # one paste-confirm dialog at a time self._resend_open = False # one resend-unscoped confirm at a time self._react_open = False # one reaction dialog at a time # The selection: the index of the highlighted message (or None when the compose # line has the focus), and the body line where it rendered, so that the frame keeps # it visible. self._selected: int | None = None self._selected_line: int | None = None # One line after the last rendered row of the selected message, so that its full # block stays visible, not only its head (refer to render_body). self._selected_end: int | None = None # The cache of the output of _render_grouped for each message. Thus a paint that # something outside the transcript causes (a key press, the 1 s idle tick, an # animation step of a spinner) renders again only the rows that changed. Refer to # _render_grouped. self._render_cache: dict | None = None # Strong references to the sends, resends, and spinner tickers that a key handler # starts. The event loop holds only a *weak* reference. Thus the garbage collector # can remove a task that nothing else names while it runs: the result is a message # that is only half transmitted, or a spinner that stops. Each task removes itself # from the set when it finishes. self._tasks: set[asyncio.Task] = set() def _spawn(self, coro) -> asyncio.Task: # noqa: ANN001 - any coroutine of this screen """Start ``coro`` on the loop and keep a reference to it until it finishes.""" task = asyncio.ensure_future(coro) self._tasks.add(task) task.add_done_callback(self._tasks.discard) return task @property def footer_hint(self) -> str: """The key hint. It shows whether a message is selected and what Enter does to it. Only the keys that do something show. ``^P paths`` must have a selected message to show the paths *of*. ``^U QR`` must have a selected message with a URL in it (refer to :meth:`_picked_urls`). ``^L map`` must have a selected message whose author has a location (refer to :meth:`_locate_target`). ``^O author`` must have a selected message whose author has a node page (refer to :meth:`_author_page`). ``^R retry failed`` must have something to retry (refer to :meth:`_retry_target`). The F-key lane uses the same rules to dim its slots. ``^K lock`` (or ``^K unlock``) must have a selected message whose author can have a lock (refer to :meth:`_lock_target`). It goes in only when there is room for it (JP, 2026-10-19: "If there's room, put it."): the hint with it must stay in 72 cells and keep each atom of the hint without it. It can make the words of an atom shorter (:data:`_SHORT_ATOMS`), as ^U, ^L, and ^O already do. When it does not go in, the status line still tells what the key did. """ hint = self._hint_with("") lock = self._lock_target() if lock is not None: atom = "^K unlock" if lock[1] else "^K lock" longer = self._hint_with(atom) # ^E react gives way to ^K, as React gives way to Lock on the Shift half of F2. if ( cell_len(longer) <= _HINT_CELLS and atom in longer.split(" · ") and _atom_set(hint) - {"^E react"} <= _atom_set(longer) ): return longer return hint def _hint_with(self, lock: str) -> str: """The key hint, with the ``lock`` atom after ``^O author`` (or no atom for ``""``).""" if self._selected is not None: code = "^U QR" if self._picked_urls() else "" where = "^L map" if self._locate_target() is not None else "" who = "^O author" if self._author_page() is not None else "" react = "^E react" if self._can_react() else "" react_first = ("^E react",) if lock else () if self._is_channel and self._retry_target() is not None: # A selected scoped message of ours can be sent again unscoped, and the hint # must name that key. The "(@mention)" of the reply and ^End are removed to # keep the line inside 72 cells (Esc still cancels the selection, and Enter # still says what it does). If the line is still too long, ↑↓ goes first, # then ^P, which the hint of each other selected message names, then ^O. # The message is ours here, so ^O only opens our own page. return _fit_hint( [ "Enter reply", "^P paths", code, where, who, lock, react, "^R resend unscoped", "↑↓ pick", "Esc cancel", ], spare=(*react_first, "↑↓ pick", "^P paths", "^O author", "^E react"), ) if self._replies: # With all the keys of a selection, the long words do not fit in 72 cells. # Then the hint uses the short words of the resend hint above. hint = _fit_hint( [ "Enter reply (@mention)", "^P paths", code, where, who, lock, react, "↑↓ pick", "^End/Esc cancel", ], spare=("↑↓ pick",), ) if cell_len(hint) <= _HINT_CELLS: return hint # ^E goes after ↑↓ when even the short words do not fit. The keys of the # selection that were here before ^E keep their places. return _fit_hint( [ "Enter reply", "^P paths", code, where, who, lock, react, "↑↓ pick", "Esc cancel", ], spare=(*react_first, "↑↓ pick", "^E react"), ) return _fit_hint( ["Enter paths", code, where, who, lock, react, "↑↓ pick", "^End/Esc cancel"], spare=(*react_first, "↑↓ pick", "^E react"), ) if self._retry_target() is not None: if self._is_channel: return "Enter send · ↑ pick a message · ^R resend unscoped · Esc back" return "Enter send · ↑ pick a message · ^R retry failed · Esc back" return "Enter send · ↑ pick a message · Esc back" # --- live updates -------------------------------------------------------- def append(self, message: ChatMessage) -> None: """Append an inbound message to the transcript and paint. A reaction goes onto the message that it reacts to, and the transcript does not move for it. """ if self._folds(message): self._session.invalidate() return self._messages.append(message) # Do not pull the viewport to the tail while the user selects 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 the status, if there is one.""" self._selected_line = None self._selected_end = None if not self._messages: lines = render_lines(Text(self._empty_text, style="muted"), width) else: # Direct threads and channel threads use the same grouped transcript, in the # style of Discord or Slack: consecutive messages from one sender share a # coloured header, with day dividers between them. Because the render is for # each message, we can also store where the selected message is (refer to # _selected_line / cursor_line). The user can select in the two types of thread. # The only difference is what Enter then does with the selection: a reply in a # channel, or the delivery paths in a direct thread. if self._selected is not None: self._selected = max(0, min(self._selected, len(self._messages) - 1)) lines = self._render_grouped(width) compose = self._compose_line(width) footer_parts: list[RenderableType] = [ Text("─" * width, style="muted"), compose, ] if self._selected is not None: footer_parts.append(self._pick_banner()) if self._status: footer_parts.append(Text(self._status, style="muted")) lines += render_lines(Group(*footer_parts), width) # Stay at the bottom: give the frame an offset that is too large, so that it clamps # the viewport to the last lines (input and latest messages). A scroll up clears # the stick. if self._stick: self.scroll = len(lines) elif self._selected_end is not None: # The selection keeps its *full* message visible, with the codes under its text. # The frame keeps only the one line that cursor_line names, the head of the # message. With only that line, a selection that moves down stops the head on # the bottom row, and the codes are below the viewport. Scroll far enough to # show the last row of the message (one extra line for the day divider pinned # at the top). The frame then moves the viewport up again only if that pushes # the head off the top. self.scroll = max(self.scroll, self._selected_end - self._scroll_viewport + 1) return lines def cursor_line(self) -> int | None: """Keep the selected reply target visible. When nothing is selected, the scroll is free (the stick controls it). """ return self._selected_line #: The text of an empty transcript. _empty_text = "No messages yet — say hello!" @property def _replies(self) -> bool: """Whether Enter on a selected message replies to it (or else opens its paths). This is true where there are many voices and an @mention says which one gets the answer, and only while there is a compose line for the reply. """ return self._multiparty and self._composing def _compose_line(self, width: int) -> RenderableType: """The compose line under the transcript: the input, with its byte limit. The byte counter is pinned to the right edge of the *last* line of the input. Thus, when the text wraps onto a second line, the counter stays in the bottom-right corner and does not follow the cursor down the wrap. Only when the last line is already full to the edge does the counter go onto its own right-aligned line. Each complete mention in the line shows as its chip, the same chip as in the transcript (refer to :class:`~meshterm.ui.compose.MentionEditor`). A chip never breaks across two lines. """ limit = self._byte_limit() input_line = self._editor.render_line( self._mention_hop, overflow_at=self._overflow_at(limit) ) whole = [(s.start, s.end) for s in input_line.spans if str(s.style) == PATH_INK] return right_aligned_tail(input_line, self._byte_counter(limit), width, whole=whole) # --- outgoing byte limit ------------------------------------------------- def _byte_limit(self) -> int: """The UTF-8 byte limit for a message in this conversation (channel or direct).""" return CHANNEL_BYTE_LIMIT if self._is_channel else DM_BYTE_LIMIT def _used_bytes(self) -> int: """The UTF-8 byte length of the current compose buffer: what counts against the limit.""" return len(self._editor.text.encode("utf-8")) def _overflow_at(self, limit: int) -> int | None: """The index of the first compose character that cannot be sent, or ``None``. With no parts, this is the first character with bytes past ``limit``. With parts, it is the first character that does not fit in the last part (refer to :func:`~meshterm.core.parts.overflow_at`). The count goes by character (not by byte), so multibyte input stays complete: an emoji or an accented letter is fully under or fully over the line, never divided in the middle of its sequence. """ return parts.overflow_at(self._editor.text, limit, self._max_parts) def _parts(self, text: str) -> list[str] | None: """The messages that ``text`` goes out as, or ``None`` when it has too many parts.""" return parts.split(text, self._byte_limit(), self._max_parts) def _byte_counter(self, limit: int) -> Text: """The inline ``used/limit`` counter: the shared compose gauge (:func:`byte_counter`). When the text goes out as more than one part, the counter starts with the number of parts, and the gauge shows the bytes of the last part, with its part number. Thus the gauge tells how much the user can add before one more part starts. """ cut = self._parts(self._editor.text.strip()) if cut is None or len(cut) < 2: return byte_counter(self._used_bytes(), limit) counter = Text(f"{len(cut)} parts · ", style="muted") counter.append_text(byte_counter(len(cut[-1].encode("utf-8")), limit)) return counter def _name(self, peer: str | None) -> str | None: """Find the contact name for a sender key prefix (exact match, 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 # --- grouped rendering --------------------------------------------------- def _render_grouped(self, width: int) -> list[str]: """Render the transcript to ANSI lines, and keep the row of the selected message. This serves direct threads and channel threads. Consecutive messages from the same sender on the same day share one coloured header, with each message body indented below it. A divider marks each new day, and a blank line separates two different sender groups. The render goes one message at a time (not as one Group). Thus we can store the body line of the selected message in :attr:`_selected_line`, so that the frame can scroll it into the viewport. The user can select in the two types of thread (refer to :meth:`handle`). Things that change nothing here cause a paint all the time: the idle tick, a key press, the ack spinner. Thus this method caches the rendered lines of each message in :attr:`_render_cache`, and for each message it puts in the cached slice instead of a new render. A message renders again only when this is necessary: * it still waits for its ack (the spinner draws its last glyph again at each tick), * it is the selected message (the selection sets its style, not the message), or * its identity or ``acked`` is different from the cache (it was appended, it was replaced in place when a pending bubble resolved, or it was removed), or * its number of reactions is different from the cache (a reaction arrived). A change of the locks draws all of the transcript again, because the sender chip of a locked contact has a padlock (refer to :meth:`_group_header`). The locks change rarely: ^K here, or the lock row of the node page of an author. Most importantly, this works for each message. It is not a cached prefix that stops at the first such row. Thus a message stays cheap to draw again, however far back it is in a long transcript. With a prefix that stops, a page up through the history causes a full render again at each key press. With this method, it does not. The grouping context (day and sender headers) is cheap to calculate again in all cases. Thus it is calculated new each time, and never taken from the cache. """ messages = self._messages total = len(messages) cache = self._render_cache spacing = self.section_spacing() locks = self._pages.locked_keys() if self._pages is not None else frozenset() if ( cache is not None and cache["width"] == width and cache["spacing"] == spacing and cache["locks"] == locks ): cached_ids = cache["ids"] cached_acked = cache["acked"] cached_selected = cache["selected"] cached_reacts = cache["reacts"] cached_boundaries = cache["boundaries"] cached_lines = cache["lines"] cache_count = len(cached_ids) else: cached_ids = cached_acked = cached_selected = cached_reacts = cached_boundaries = () cached_lines = [] cache_count = 0 lines: list[str] = [] # Store each day divider as a sticky block of one row. Thus, when the divider of # the top visible message scrolls off, it is pinned to the top row again. This is # the same base Screen.sticky_block that the conversation picker uses for its # section headings. A day has nothing to say except its date, so the block is only # the divider. The heading of a select list can also have its description. self._sticky_headers = [] ids: list[int] = [] ackeds: list[bool | None] = [] selecteds: list[bool] = [] reacts: list[int] = [] boundaries: list[int] = [] prev_group: tuple[bool, str] | None = None prev_day = None # The first message of the group of the selected message. Its slice has the author # chip of the group, which is white while the selection is in the group. picked_head = self._picked_head() for idx in range(total): message = messages[idx] stamp = message.created_at.astimezone() day = stamp.date() sender, body = self._sender_and_body(message) # Group by (are-we-the-sender, display name), not by the name alone. Thus our # messages never join with those of a remote sender that has the same name. group = (message.outbound, sender) new_day = day != prev_day selected = idx == self._selected # A slice that the selection changes renders again, now and on the next paint # after the selection leaves it: the selected message, and the head of its group. marked = selected or idx == picked_head pending = message.acked is None and message.outbound and not message.is_channel react_count = self._fold.count(message) if self._fold is not None else 0 reusable = ( not marked and not pending and idx < cache_count and not cached_selected[idx] and cached_ids[idx] == id(message) and cached_acked[idx] == message.acked and cached_reacts[idx] == react_count ) # A blank line closes the messages of an author and the messages of a day, where # the preferences ask for it (:meth:`~meshterm.ui.tui.screen.Screen.section_spacing`, # the same test as the section headings of a list). It is the first line of the # slice of the message that starts the next group, so a cached slice brings it # back with it. gap = spacing and bool(lines) and (new_day or group != prev_group) if reusable: start = cached_boundaries[idx - 1] if idx > 0 else 0 slice_lines = cached_lines[start : cached_boundaries[idx]] if new_day: at = int(gap) # the divider comes after the gap self._sticky_headers.append((len(lines) + at, [slice_lines[at]])) lines += slice_lines else: if gap: lines.append("") if new_day: # A day divider is the section heading of this transcript: it pins to the # top row while its day scrolls under it, and ^PgUp/^PgDn step by it. # Thus it has the heading style of the app (``── Label ──`` in the bold # ``heading`` grey, refer to menus.section_heading). The rule above the # compose line is plain ``muted``, because that rule is chrome. divider = render_lines( Text(f"── {stamp:%a} {stamp:%b} {stamp.day} ──", style="heading"), width ) self._sticky_headers.append((len(lines), [divider[0]])) lines += divider if new_day or group != prev_group: header = self._group_header(sender, message, picked=idx == picked_head) lines += render_lines(header, width) if selected: # The first message has the head of the transcript: nothing comes before # its day divider. Thus its selection anchors at line 0, not at its own # body. When the anchor was on the body, the viewport at the very top # was scrolled down two lines: the divider was only pinned, the sender # chip was not visible, and the ↑ of the title bar was lit over a # transcript with nothing above it. self._selected_line = 0 if idx == 0 else len(lines) lines += self._body_lines(body, message, width, selected=selected) if selected: self._selected_end = len(lines) prev_group, prev_day = group, day ids.append(id(message)) ackeds.append(message.acked) selecteds.append(marked) reacts.append(react_count) boundaries.append(len(lines)) self._render_cache = { "width": width, "spacing": spacing, "locks": locks, "ids": ids, "acked": ackeds, "selected": selecteds, "reacts": reacts, "lines": lines, "boundaries": boundaries, } return lines def _sender_and_body(self, message: ChatMessage) -> tuple[str, str]: """Return the display sender and the clean body of a message. Our messages are ``you``. An inbound channel message has a ``Name: `` prefix, which we move into the sender (``·`` when there is no prefix). An inbound direct message gets the sender from the resolved contact name (or the raw key, or ``?``). """ if message.outbound: return "you", message.text if message.is_channel: name, body = split_channel_sender(message.text) return (name or "·"), body return (self._name(message.peer) or message.peer or "?"), message.text def _picked_head(self) -> int | None: """The index of the first message of the group of the selected message, or ``None``. A group is a run of messages from one sender on one day, under one author chip (refer to :meth:`_render_grouped`). The chip is drawn with the first message of the run, so that message is the one to draw again when the selection is in the run. """ selected, messages = self._selected, self._messages if selected is None or not 0 <= selected < len(messages): return None def group_of(message: ChatMessage) -> tuple: day = message.created_at.astimezone().date() return (day, message.outbound, self._sender_and_body(message)[0]) group = group_of(messages[selected]) head = selected while head > 0 and group_of(messages[head - 1]) == group: head -= 1 return head def _group_header(self, sender: str, message: ChatMessage, *, picked: bool = False) -> Text: """Build the sender header that starts a group: the name, drawn as a chip. The chip makes a *label* different from the prose under it. A name alone above its messages then reads as a tag, not as a coloured word that a person typed. It is the chip of a path line (refer to :func:`~meshterm.ui.widgets.name_chip`). Thus a sender looks exactly like that node as a hop in a route. Our messages are the exception. They go under our name, as ``★ Name`` on a chip in white (refer to :func:`~meshterm.ui.widgets.self_chip`). A route shows our end as the bare ``★``, but a chat is a conversation between names. Our chip comes from the direction of the message (``outbound``), not from the ``"you"`` label. Thus a remote sender with the name ``you`` still gets the hue of its key, and does not look like our node. The sender is the origin of the route of each message under it. Thus the chip closes on the point, the mark of a route that goes on. An ``@mention`` in a body is the other end, and it opens on the notch (refer to :meth:`_render_mentions`). A sender that is a locked contact has the padlock before its name, in the place of the ``★`` of our chip (JP, 2026-10-19). Thus the chip shows the state that ^K toggles. With ``picked``, the selected message is in this group, and the label of the chip is white, as the text of that message is (refer to :func:`~meshterm.ui.widgets.picked_hop`, JP, 2026-10-19). Our chip already has a white label. """ if message.outbound: return self_chip(self._self_name, self._self_key, role=ORIGIN) # ``·`` is a channel line that arrived with no sender prefix: there is no key to use. key = None if sender == "·" else self._header_key(sender, message) pages = self._pages locked = key is not None and pages is not None and pages.lock_state(pages.find(key)) is True return author_chip(sender, key, locked=locked, picked=picked) def _header_key(self, sender: str, message: ChatMessage) -> str | None: """The key that gives the colour of the sender chip of a group: the key of the name. This is a hook, not a call, because a room knows more than a name. Each post has the key prefix of its author, which gives the colour of the chip with no name lookup (refer to :class:`~meshterm.ui.room.RoomScreen`). """ return self._sender_key(sender) def _body_lines( self, body: str, message: ChatMessage, width: int, *, selected: bool = False ) -> list[str]: """Render one message to ANSI lines, with its own time and a hanging indent. The timestamp of each message is here (not on the group header), so that each message shows the time when it was sent, also when one sender has many messages in a group. Without this, a run of messages from the same sender seems to share one time. A long body wraps with a hanging indent, so that the next lines align under the body and not under the timestamp gutter. When ``selected`` is true, the line is marked as the reply target (the same as the ``❯`` pointer and the highlight of the select screen). A message that mentions our name has a mark at the left of its time: the white ``★`` of our node (refer to :func:`~meshterm.ui.widgets.mention_mark`, JP, 2026-10-13). The chip of the mention in the body has our white label, but no star (refer to :func:`~meshterm.ui.widgets.mention_hop`). Thus the mark of a message that names us is in one column, where the eye finds it when it goes down the transcript. Each time has the two cells of that column in front of it, so that all the times stay in one column too. A URL is never cut where the screen can hold it. A URL that is too long for the body lane moves out to the column after the ``❯``, on its own line (refer to :func:`~meshterm.ui.tui.render.render_hanging`). This is because a terminal opens a link, and a user copies one, only while it is on one line. Where the platform draws them (:data:`_URL_CODES`, the PicoCalc), each URL in the body gets a QR code under it. The codes are side by side, at the indent of the body. Thus the user can open a link from the handheld on a phone, and does not have to type it. """ stamp = message.created_at.astimezone() prefix = Text() prefix.append("❯ " if selected else " ", style="cursor" if selected else None) if self._names_us(body): prefix.append_text(mention_mark(self._self_key)) prefix.append(" ") else: prefix.append(" ") prefix.append(f"{stamp:%H:%M} ", style="cursor" if selected else "muted") body_text = self._body_text(body, message, selected=selected) whole = [m.span() for m in _WORD.finditer(body_text.plain) if links.urls(m.group())] # A mention is a chip, and a chip never breaks across two lines. whole += [(s.start, s.end) for s in body_text.spans if str(s.style) == PATH_INK] lines = render_hanging( prefix, body_text, width, indent=prefix.cell_len, whole=whole, gutter=_URL_GUTTER ) urls = links.urls(body) if _URL_CODES else [] if urls: lines += render_lines(qr_strip(urls, width, indent=prefix.cell_len), width) badges = self._fold.badges(message) if self._fold is not None else [] if badges: row, runs = self._badge_text(badges, selected=selected) lines += render_hanging( Text(" " * prefix.cell_len), row, width, indent=prefix.cell_len, whole=runs ) return lines def _badge_text( self, badges: list[reactions.Badge], *, selected: bool ) -> tuple[Text, list[tuple[int, int]]]: """The reactions on one message, as one line under its body, and the runs to keep whole. Each emoji has its count after it. The count is ``accent`` when our node is one of the nodes that reacted, else ``muted``. On the selected message, the names of the nodes take the place of the count, each in the hue of its node, and our node is the white ``you``. Where the font has no emoji, an emoticon stands in for each default reaction (refer to :func:`~meshterm.ui.theme.emoticon`). """ text = Text() runs: list[tuple[int, int]] = [] for index, badge in enumerate(badges): if index: text.append(" ") start = len(text) text.append(emoticon(badge.emoji)) text.append(" ") if not selected: count = f"{_COUNT_MARK}{len(badge.reactors)}" text.append(count, style="accent" if badge.ours else "muted") runs.append((start, len(text))) continue for number, reactor in enumerate(badge.reactors): if number: text.append(", ", style="muted") if reactor.ours: text.append("you", style="you") else: name = reactor.name or self._peer_label key = self._peer_key if reactor.name is None else self._sender_key(name) text.append(name, style=name_style(name, key or None)) if not number: runs.append((start, len(text))) # the emoji never ends a line alone return text, runs def _body_text(self, body: str, message: ChatMessage, *, selected: bool) -> Text: """Build the styled body of a message: mentions in colour, then the glyphs at the end.""" text = self._render_mentions(body, selected=selected) if message.outbound: # A direct message shows its own delivery (spinner → ✓/✗, and a retry is # possible). A channel broadcast has no ack. Thus mark only a channel message # that did not leave the companion. if message.is_channel: if message.acked is False: text.append(" ⚠ no ack", style="warn") text.append_text(self._scope_note(message)) else: text.append(" ") text.append_text(self._delivery_glyph(message.acked)) if message.snr is not None: text.append(f" {message.snr:+.0f} dB", style=snr_style(message.snr)) return text def _scope_note(self, message: ChatMessage) -> Text: """A muted note after a sent message that went out under a scope not in the title. This note shows only in a channel whose title names a scope, and only for the exceptions to that scope: an unscoped resend, or a message sent before the scope was set or changed. On each other line, a note only repeats the title. In a channel with no scope of its own, the default scope of its messages belongs to the device. It is not news to show on each line. """ if not self._scope or not message.scope or message.scope == self._scope: return Text() if message.scope == WILDCARD: return Text(" · unscoped", style="muted") return Text(f" · scope {message.scope}", style="muted") def _render_mentions(self, body: str, *, selected: bool) -> Text: """Render body text, and change each ``@[Name]`` token to the chip of that node. The reply flow puts an ``@[Name]`` token in the compose line (refer to :meth:`_begin_reply`). Here it shows as the name in a chip of the hue of that node. Thus the user sees that a mention belongs to the node that it names. The text around the mentions keeps the base style of the line (``cursor`` when the message is the selected reply target, or else no style). A mention of our own name has our chip, in white, and the line has our white ``★`` at the left of its time (refer to :meth:`_body_lines`). Thus a message that names us is easy to see. The chip opens on the notch, because the node that a mention names is the destination of a route that came from the sender (refer to :meth:`_group_header`, whose chip closes on the point). The chip has no ``@``, because the chip already says that the name is a mention (JP, 2026-10-13). Arrow mode has no chip, so there a mention keeps its ``@`` (refer to :func:`~meshterm.ui.widgets.mention_hop`). On the selected message, the label of each chip is white, as the text around it is (refer to :func:`~meshterm.ui.widgets.picked_hop`, JP, 2026-10-19). """ base = "cursor" if selected else None text = Text() pos = 0 for match in MENTION.finditer(body): if match.start() > pos: text.append(body[pos : match.start()], style=base) hop = self._mention_hop(match.group(1)) if selected: hop = picked_hop(hop) # the chips turn white with the text around them text.append_text(hop_chip(hop, role=DESTINATION)) pos = match.end() if pos < len(body): text.append(body[pos:], style=base) return text def _mention_hop(self, name: str) -> PathHop: """The hop of a mention of ``name``: in the transcript, and in the compose line. The two use this one method, so a chip in the compose line is the chip that the message will show after the send. A mention of our name has our key and the chip with the white label. Each other name has the key that the names of the senders give (refer to :meth:`_sender_key`). """ if self._self_name and name == self._self_name: return mention_hop(name, self._self_key, you=True) return mention_hop(name, self._sender_key(name, mention=True)) def _names_us(self, body: str) -> bool: """Whether ``body`` has a mention of our name (it gets a mark before its time).""" if not self._self_name: return False return any(match.group(1) == self._self_name for match in MENTION.finditer(body)) def _delivery_glyph(self, acked: bool | None) -> Text: """Map the ``acked`` state of an outbound direct message to the mark at its end. A message that still waits for its ack (``acked is None``) shows the current animation step of the spinner. :meth:`_spin_while` animates it while the send is in flight. A resolved message shows the delivered ``✓`` (ok) or the unacknowledged ``✗`` (err). """ if acked is None: return self._spinner.text() glyph, style = _DELIVERED if acked else _FAILED return Text(glyph, style=style) def _pick_banner(self) -> Text: """A one-line note above the input while a message is selected. In a channel, it starts with the reply (what Enter does there). In a direct chat, it starts with the paths view (what Enter does there). The two say what the selection is for. Thus the user never sees a highlight that has no explanation. """ message = self._messages[self._selected] sender, _ = self._sender_and_body(message) who = "this message" if message.outbound or sender == "·" else sender if self._replies: return Text( f"↩ Enter to reply to {who} with an @mention · End to cancel", style="accent", ) return Text("Enter to see the paths this message took · End to cancel", style="accent") def _sender_key(self, sender: str, *, mention: bool = False) -> str | None: """The key of a sender name, or ``None`` when nothing gives a key for it. The contacts first, then the names that the recorder stored. In a *direct* thread, a sender label with no key falls back to the key of the peer, because the only two persons in the conversation are we and the peer. A mention names any node, so that fallback must not apply to a mention. Thus a mention that is not resolved stays grey in a direct chat, exactly as it does in a channel. A name with no key is drawn in the *node* grey, never in ``muted`` (the chip does this, refer to :func:`~meshterm.ui.widgets.name_chip`). An unidentified sender is content that you can still do actions on (select the message, open its paths), not chrome. On the console, ``muted`` is the dark slot and ``node.unknown`` is the light one (JP, 2026-08-12). The sender chip (:meth:`_group_header`) and the mention chips (:meth:`_render_mentions`) share this method. Thus the two cannot disagree about the identity of a name. """ key = self._key_of(sender) if not key and not self._multiparty and not mention: key = self._peer_key or None return key # --- input --------------------------------------------------------------- def handle(self, action: str, data: str = "") -> None: """Dispatch a key: select messages and do actions on them, edit the compose line, or leave. The two types of chat share one model. When no message is selected, Enter sends, and typing edits the compose line. An amber dialog asks the user to confirm a paste (Ctrl-V, or the bracketed paste of a terminal) before the paste goes into the line (refer to :meth:`_begin_paste`). ↑ selects the newest message. The selection then moves with ↑↓, PgUp/PgDn (one screen), Ctrl+Home (the very first message), and Ctrl+PgUp/PgDn (day dividers), and the viewport moves with it. Enter on a selected message starts a reply with an ``@mention`` in a channel, and opens the delivery paths in a direct chat. In the two types of chat, ^P opens the paths of the selected message, ^U the QR codes of its links, ^L the map on its author, and ^E the reaction dialog. ^K locks or unlocks its author. These keys do nothing when no message is selected. ^End (or a move past the newest message) goes back to the compose line. The first Esc clears the selection, and the second Esc leaves the screen. """ if action == "enter": if self._selected is not None: if self._replies: self._begin_reply() else: self._open_paths(self._selected) else: self._submit() elif action == "paste": self._begin_paste(data) elif action == "paths": self._open_paths(self._selected) elif action == "url_code": self._open_codes() elif action == "locate": self._open_map() elif action == "author_page": self._open_author_page() elif action == "react": self._begin_react() elif action == "lock": self._toggle_lock() elif action == "retry": if self._is_channel: self._begin_resend_unscoped() else: self._retry() elif action == "escape": if self._selected is not None: self._clear_selection() # the first Esc clears the selection, the next leaves self._session.invalidate() else: self.resolve(CANCEL) elif action == "up": self._move_selection(-1) elif action == "pageup": self._move_selection(-self._page_step) elif action == "down": self._move_selection(1) elif action == "pagedown": self._move_selection(self._page_step) elif action == "ctrl_home": if self._messages: self._selected = 0 self._stick = False elif action == "ctrl_pageup": self._select_section(-1) elif action == "ctrl_pagedown": self._select_section(1) elif action == "ctrl_end": self._clear_selection() self._stick = True # jump back to the live tail and the compose line else: if self._composing and self._editor.edit(action, data): # An edit of the compose line puts the focus there. Nothing stays selected. self._status = "" # an edit clears the "too long" notice self._clear_selection() self._stick = True def _open_paths(self, index: int | None) -> None: """Float the delivery-paths view for the selected message (one dialog at a time). ``index`` is the selection. Thus ``None`` (nothing selected) opens nothing. A path is a route that one *specific* message walked. When the code guessed the latest message, it showed the paths of the message that was at the bottom of the transcript at that time. """ if index is None or not self._messages or self._paths is None or self._paths_open: return message = self._messages[max(0, min(index, len(self._messages) - 1))] self._paths_open = True async def run() -> None: try: await self._paths(message) finally: self._paths_open = False self._session.invalidate() self._session.run_detached(run()) def _locate_target(self) -> MapMarker | None: """The marker of the author of the selected message, or ``None`` when ^L does nothing. The author is the node whose key gives the colour of the chip of the message (refer to :meth:`_header_key`). Our own message has our node as its author. A message with no sender, or an author that has no location, has no marker. """ if self._locator is None or self._selected is None or not self._messages: return None message = self._messages[max(0, min(self._selected, len(self._messages) - 1))] if message.outbound: return self._locator.you() sender, _body = self._sender_and_body(message) if sender == "·": return None return self._locator.find(self._header_key(sender, message)) def _open_map(self) -> None: """Open the map on the author of the selected message (^L). One map at a time. The map is a screen, and it goes on the stack above the chat. Esc on the map comes back to the chat, with the same message selected. """ target = self._locate_target() if target is None or self._locator is None or self._map_open: return locator = self._locator self._map_open = True async def run() -> None: try: await locator.show(target) finally: self._map_open = False self._session.invalidate() self._session.run_detached(run()) def _author_page(self) -> Callable[[], Awaitable[None]] | None: """The node page that ^O opens for the selected message, or ``None`` when there is none. The page belongs to the author of the message: our node for a message of ours, or the contact that the key of the author finds (refer to :meth:`_header_key`, the same key that gives the colour of the sender chip). A message with no sender, or an author that is not a contact, has no page. The hint and the F-key lane ask this at each paint, so that they name ^O only when it acts. """ pages = self._pages if pages is None or self._selected is None or not self._messages: return None message = self._messages[max(0, min(self._selected, len(self._messages) - 1))] if message.outbound: return lambda: pages.show(None) contact = self._author_contact() return None if contact is None else (lambda: pages.show(contact)) def _author_contact(self) -> Contact | None: """The contact that wrote the selected message, or ``None``. The key of the author finds it (refer to :meth:`_header_key`, the same key that gives the colour of the sender chip). A message of ours, a message with no sender, and an author that is not a contact have none. """ pages = self._pages if pages is None or self._selected is None or not self._messages: return None message = self._messages[max(0, min(self._selected, len(self._messages) - 1))] if message.outbound: return None sender, _body = self._sender_and_body(message) if sender == "·": return None return pages.find(self._header_key(sender, message)) def _lock_target(self) -> tuple[Contact, bool] | None: """The author that ^K locks or unlocks, and whether it is locked now. ``None`` when ^K does nothing: no selected message, a message of ours, an author that is not a contact, or a contact that cannot have a lock (refer to :meth:`~meshterm.ui.node_detail_screen.ContactPages.lock_state`). The hint and the F-key lane ask at each paint, and the store answers from memory. """ contact = self._author_contact() if contact is None or self._pages is None: return None locked = self._pages.lock_state(contact) return None if locked is None else (contact, locked) def _toggle_lock(self) -> None: """Lock the author of the selected message, or unlock it when it is locked (^K). There is no confirm, because the next ^K reverses it, as the row on the node page does. The hint atom and the lane chip change to their opposite. The status line also tells what the key did, because the hint has no room for the atom on each message. """ target = self._lock_target() if target is None or self._pages is None: return contact, locked = target self._pages.set_locked(contact, not locked) name = contact.name or contact.key_prefix self._status = f"Unlocked {name}" if locked else f"Locked {name} — no sweep archives it" self._session.invalidate() def _open_author_page(self) -> None: """Open the node page of the author of the selected message (^O). One page at a time. The page is a screen, and it goes on the stack above the chat. Esc on the page comes back to the chat, with the same message selected. """ page = self._author_page() if page is None or self._page_open: return self._page_open = True async def run() -> None: try: await page() finally: self._page_open = False self._session.invalidate() self._session.run_detached(run()) def _picked_urls(self) -> list[str]: """The URLs in the body of the selected message, in order. None if nothing is selected.""" if self._selected is None or not self._messages: return [] message = self._messages[max(0, min(self._selected, len(self._messages) - 1))] return links.urls(self._sender_and_body(message)[1]) def _open_codes(self) -> None: """Show the links of the selected message as QR codes on the share screen (^U). This is the share screen of a channel or of a contact card (:class:`~meshterm.ui.qr.QrScreen`), full-frame. Thus a phone can read a link from here and open it. Many links are on one screen that ←→ step through, in the order of the message. The same as ^P, it acts on the *selection*: a link belongs to one message, and when nothing is selected, the screen cannot know which message. One share screen at a time. """ urls = self._picked_urls() if not urls or self._codes_open: return self._codes_open = True async def run() -> None: try: await self._session.run_screen(QrScreen(*urls, title="Links")) finally: self._codes_open = False self._session.invalidate() self._session.run_detached(run()) def _begin_paste(self, data: str) -> None: """Confirm a clipboard paste on an amber dialog, then insert it into the compose line. A chat message goes out over the air. A paste can be much larger than a key press, or it can have content that the user did not want to broadcast. Thus a paste must go through a Cancel/Paste confirm (the amber ``danger`` tier: disruptive, not a loss of data), and does not go directly into the line. Newlines and control characters become spaces (a message is one line). An empty result never opens the dialog. One paste dialog at a time, scheduled from the key handler, the same as the paths view. """ if self._paste_open or not self._composing: return clean = "".join(ch if ch.isprintable() else " " for ch in data).strip() if not clean: return self._paste_open = True count = len(clean) async def run() -> None: try: confirmed = await self._session.button_dialog( Text( f"Paste {count} character{'s' if count != 1 else ''} into your message?", style="warn", ), [("Cancel", False), ("Paste", True)], title="Paste", default=1, border_style="warn", footer_hint="←→ choose · Enter select · Esc cancel", ) finally: self._paste_open = False if confirmed: # Put it in the compose line: clear the selection, go back to the tail, and # insert the run through the shared editor. If the result is over the limit, # the byte gauge marks it, exactly as it marks typing past the limit. self._clear_selection() self._stick = True self._editor.edit("text", clean) self._status = "" self._session.invalidate() self._spawn(run()) def _move_selection(self, delta: int) -> None: """Move the reply selection by ``delta`` messages (negative = toward older). The selection starts from the compose line only on a move up (``delta < 0``). A move past the newest message clears the selection and stays at the tail again. """ 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 _select_section(self, direction: int) -> None: """Move the reply selection to the first message of the previous or next day. This is the day jump of the transcript (Ctrl+PageUp/PageDown), for the selection. Down moves to the first message of the next day (or to the tail when there is no later day). Up moves to the first message of the current day, or to the first message of the previous day when the selection is already on a first message. """ if not self._messages: return starts = self._day_start_indices() current = self._selected if self._selected is not None else len(self._messages) - 1 if direction > 0: target = next((s for s in starts if s > current), None) if target is None: self._clear_selection() self._stick = True return else: governing = max((s for s in starts if s <= current), default=0) target = ( governing if governing < current else max((s for s in starts if s < current), default=0) ) self._selected = target self._stick = False def _day_start_indices(self) -> list[int]: """The indexes of the messages that start a new local day: the transcript dividers.""" starts: list[int] = [] prev_day = None for i, message in enumerate(self._messages): day = message.created_at.astimezone().date() if day != prev_day: starts.append(i) prev_day = day return starts def _clear_selection(self) -> None: """Clear the reply selection (the focus goes back to the compose line).""" self._selected = None self._selected_line = None def _begin_reply(self) -> None: """Put an ``@mention`` of the sender of the selected message at the cursor. The mention goes in at the cursor of the compose line, and the text that the user already typed stays. Thus a message can name more than one node: select a message, press Enter, select another, and press Enter again. A space comes before the mention when it would touch the word in front of it, and after it when nothing follows or a word would touch it. The mention is a chip from that moment, and the cursor goes after it (refer to :class:`~meshterm.ui.compose.MentionEditor`). A message of ours, or one with no sender, gives no name to mention. There the line gets a bare ``@``, and the user types the name. """ message = self._messages[self._selected] sender, _ = self._sender_and_body(message) named = not message.outbound and sender != "·" text, cursor = self._editor.text, self._editor.cursor before = " " if cursor > 0 and not text[cursor - 1].isspace() else "" if named: after = " " if cursor == len(text) or not text[cursor].isspace() else "" self._editor.insert(f"{before}@[{sender}]{after}") else: self._editor.insert(f"{before}@") self._clear_selection() self._stick = True self._session.invalidate() def _submit(self) -> None: """Send the current input line as a message (scheduled from the key handler). A line that is too long for one packet goes out as parts, each with its part number (refer to :mod:`meshterm.core.parts`), when the ``message_parts_max`` preference permits more than one part. A line that is too long for all the parts is refused. The buffer stays the same (so that the user can make it shorter), and the status shows why, the same as the red overflow that the compose bar already shows. """ text = self._editor.text.strip() if not text or self._sending: return cut = self._parts(text) if cut is None: if self._max_parts < 2: over = self._used_bytes() - self._byte_limit() self._status = f"Too long by {over} byte{'s' if over != 1 else ''} — trim to send." else: self._status = f"Too long for {self._max_parts} parts — trim to send." self._session.invalidate() return self._editor = MentionEditor() self._sending = True self._stick = True if self._is_channel: self._status = "sending…" self._session.invalidate() self._spawn(self._send_channel(cut)) else: self._status = "" self._session.invalidate() self._spawn(self._send_direct(cut)) def _sending_part(self, index: int, total: int) -> None: """Show which part goes out now, when a message has more than one.""" if total > 1: self._status = f"sending part {index} of {total}…" self._session.invalidate() def _keep_unsent(self, cut: list[str], first: int) -> None: """Put the text of the parts that did not go out back in the compose line. ``first`` is the index in ``cut`` of the first part that did not go out. The text goes back without its part numbers, so that Enter cuts it again. If the user typed something new in the meantime, the line stays as it is, and the text is lost. """ if self._editor.text or first >= len(cut): return total = len(cut) rest = " ".join(parts.unnumbered(cut[i], i + 1, total) for i in range(first, total)) self._editor = MentionEditor(rest) async def _send_channel(self, cut: list[str]) -> None: """Broadcast each part of a channel message, and append each when the device takes it. Between two parts, the send waits ``part_gap_s``, so that the floods do not go out as a burst. If a part fails, the parts after it do not go out, and their text goes back in the compose line. """ sent = 0 try: for i, text in enumerate(cut): if i: await asyncio.sleep(self._part_gap_s) self._sending_part(i + 1, len(cut)) message = await self._send(text) sent += 1 if message is not None: self._admit(message) self._session.invalidate() self._status = "" except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open self._status = f"send failed: {exc}" self._keep_unsent(cut, sent) finally: self._sending = False self._stick = True self._session.invalidate() async def _spin_while(self, coro: Awaitable[Any]) -> Any: """Await ``coro`` while the delivery spinner animates on the message in flight. A background timer moves the shared spinner forward and paints every :func:`~meshterm.ui.tui.spinner.spinner_interval` seconds (the spin rate of the platform). Thus the glyph at the end of a pending message spins until the ack resolves. Before the method returns, it always cancels the timer (and awaits it, so that it cannot stay after the send as a stray pending task). The spinner is only for the look. Thus the method ignores each problem in the animation, so that the problem cannot break the send. """ self._spinner.reset() async def animate() -> None: while True: await asyncio.sleep(spinner_interval()) self._spinner.tick() self._session.invalidate() ticker = self._spawn(animate()) try: return await coro finally: ticker.cancel() try: await ticker except asyncio.CancelledError: pass except Exception: # noqa: BLE001 - a spinner problem must never break a send pass async def _send_direct(self, cut: list[str]) -> None: """Send each part of a direct message, and show each as an optimistic bubble. Each part shows as a pending bubble while its delivery is in flight. On success, the stored message replaces the bubble. The stored message has its resolved ``acked`` state and its row id, so that the user can retry a ✗ later in place. A part goes out when the part before it is acknowledged or its wait ends. A part with no acknowledgement can still have arrived, so the parts after it go out too, and ^R retries each ✗ in turn. A hard failure (the companion refuses the send) removes the bubble, shows the error inline, and stops the parts after it. Their text goes back in the compose line. """ sent = 0 try: for i, text in enumerate(cut): self._sending_part(i + 1, len(cut)) pending = ChatMessage(text=text, outbound=True, created_at=utcnow()) self._messages.append(pending) self._session.invalidate() try: message = await self._spin_while(self._send(text)) except Exception: self._discard(pending) raise if message is None: # nothing went out (the contact was not restored) self._discard(pending) self._keep_unsent(cut, sent) break sent += 1 self._swap(pending, message) self._status = "" except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open self._status = f"send failed: {exc}" self._keep_unsent(cut, sent) finally: self._sending = False self._stick = True self._session.invalidate() def _retry_target(self) -> ChatMessage | None: """The message that ^R sends again, or ``None`` when there is nothing to retry. In a direct chat, this is the newest outbound message that went out and was never acknowledged. In a channel, nothing is ever acknowledged. There, it is a message that *we* sent under a region, and ^R sends it again unscoped: the **selected** message when a message is selected (the user pointed at it), or else the newest message that we sent. With a selection, only the selection counts. If the selected message is not a scoped message of ours, ^R offers nothing, and does not go past it to another message without a sign. With no selection, only the newest counts. When its unscoped copy has gone, that copy is the newest, and nothing remains to offer. In the two cases, this is true only while a resend can start: no send is already in flight, and a resend path is connected. """ if self._is_channel: if self._sending or self._resend_unscoped is None: return None if self._selected is not None: candidate: ChatMessage | None = self._messages[self._selected] else: candidate = next((m for m in reversed(self._messages) if m.outbound), None) if ( candidate is None or not candidate.outbound or not candidate.scope or candidate.scope == WILDCARD ): return None return candidate if self._sending or self._resend is None: return None return next( (m for m in reversed(self._messages) if m.outbound and m.acked is False), None, ) def _retry(self) -> None: """Try again to deliver the most recent unacknowledged direct message (Ctrl-R).""" target = self._retry_target() if target is None: return self._sending = True target.acked = None # back to the spinner while the retry is in flight self._status = "retrying…" self._stick = True self._session.invalidate() self._spawn(self._resend_message(target)) def _begin_resend_unscoped(self) -> None: """Confirm, then send a scoped channel message again, unscoped (^R). The dialog is amber (``danger``). Nothing is lost, but each repeater that permits an unscoped flood relays it: all the mesh that the scope kept the message out of. Thus it is a choice made on purpose, and the dialog names the region that the message leaves. The resend is a new message on the air (a channel message has no identity to deliver again), so it is appended as a new message. A radio that cannot send unscoped refuses before anything goes out, and the status line says why. """ target = self._retry_target() if target is None or self._resend_open: return self._resend_open = True preview = target.text if len(target.text) <= 32 else target.text[:31] + "…" async def run() -> None: try: confirmed = await self._session.button_dialog( Text( f"Send “{preview}” again, unscoped? Every repeater that relays " f"unscoped floods will carry it, not only those in {target.scope}.", style="warn", ), [("Cancel", False), ("Resend", True)], title="Resend unscoped", default=1, border_style="warn", footer_hint="←→ choose · Enter select · Esc cancel", ) finally: self._resend_open = False if not confirmed or self._sending or self._resend_unscoped is None: self._session.invalidate() return self._sending = True self._status = "resending unscoped…" # The resend goes to the tail as a new message. The selection that chose it is # done. self._clear_selection() self._stick = True self._session.invalidate() try: message = await self._resend_unscoped(target) if message is not None: self._admit(message) self._status = "" except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open self._status = f"resend failed: {exc}" finally: self._sending = False self._stick = True self._session.invalidate() self._spawn(run()) async def _resend_message(self, message: ChatMessage) -> None: """Run a retry to its end, and refresh the delivery state of the message in place.""" assert self._resend is not None try: await self._spin_while(self._resend(message)) except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open message.acked = False self._status = f"retry failed: {exc}" else: self._status = "" if message.acked else "still no ack — ^R to retry" finally: self._sending = False self._stick = True self._session.invalidate() def _swap(self, old: ChatMessage, new: ChatMessage) -> None: """Replace an optimistic bubble with its stored message, in place. The stored message is the one that a reaction can name, because it has the timestamp that went out. Thus the fold takes it here, and not the bubble. """ if self._folds(new): self._discard(old) return try: self._messages[self._messages.index(old)] = new except ValueError: # pragma: no cover - the bubble is always still there self._messages.append(new) def _folds(self, message: ChatMessage) -> bool: """Give ``message`` to the fold. ``True`` for a reaction, which the transcript omits.""" return self._fold is not None and self._fold.take(message) def _admit(self, message: ChatMessage) -> None: """Append a message that we sent, unless it is a reaction (the fold shows that one).""" if not self._folds(message): self._messages.append(message) # --- reactions ------------------------------------------------------------- def _react_target(self) -> tuple[ChatMessage, str | None, str, int] | None: """The selected message, and the author, body, and timestamp that a reaction names it by. ``None`` when nothing is selected, or when a reaction cannot name the selection: a room, a message that is still in flight, or a channel message with no author (a line with no ``Name: `` prefix, or one of ours when our name is not known). """ if self._fold is None or self._selected is None or not self._messages: return None message = self._messages[min(self._selected, len(self._messages) - 1)] named = self._fold.target_of(message) if named is None: return None author, body, timestamp = named if self._is_channel and not author: return None return message, author, body, timestamp def _can_react(self) -> bool: """Whether ^E can start a reaction: a selection that a reaction can name, and no send.""" return not self._sending and self._composing and self._react_target() is not None def _begin_react(self) -> None: """Ask which emoji, then send the reaction to the selected message (^E). The dialog is a question, so it floats as a box over the chat. The emoji of the ``reaction_emoji`` preference come first, then each other emoji (refer to :func:`emoji_catalog`). Each row has the name of its emoji, so a typed word such as ``fire`` or ``heart`` filters the list to the emoji that it names. A row that our node already sent to this message says so. A second copy of it would only count one time, so the selection of that row sends nothing. The reaction goes out as a usual message in the conversation. When it is stored, the fold puts it under its target (refer to :meth:`_admit`). """ target = self._react_target() if target is None or self._sending or self._react_open or self._fold is None: return message, author, body, timestamp = target fold = self._fold self._react_open = True sender, _ = self._sender_and_body(message) who = "your message" if message.outbound else sender async def run() -> None: from .menus import menu_rows, section_heading def row(emoji: str) -> tuple[str, str, str]: name = reactions.emoji_name(emoji) sent = fold.reacted(message, emoji) return emoticon(emoji), f"{name} · sent" if sent else name, emoji mine = list(self._reaction_set) others = [emoji for emoji in emoji_catalog() if emoji not in mine] rows = menu_rows([row(emoji) for emoji in mine + others]) items = [ section_heading("Your reactions"), *rows[: len(mine)], section_heading("All emoji"), *rows[len(mine) :], ] try: emoji = await self._session.select( f"React — {who}", items, filterable=True, floating=True ) finally: self._react_open = False if not emoji or self._sending: self._session.invalidate() return if fold.reacted(message, emoji): self._status = "you already reacted with that" self._session.invalidate() return if self._is_channel: text = reactions.channel_text(emoji, author or "", body, timestamp) else: text = reactions.direct_text(emoji, body, timestamp) self._sending = True self._status = "reacting…" self._session.invalidate() try: sent_message = await self._send(text) if sent_message is not None: self._admit(sent_message) no_ack = not self._is_channel and sent_message.acked is False self._status = "reaction sent · no ack" if no_ack else "" else: self._status = "" except Exception as exc: # noqa: BLE001 - show the error inline, keep the chat open self._status = f"reaction failed: {exc}" finally: self._sending = False self._session.invalidate() self._spawn(run()) def _discard(self, message: ChatMessage) -> None: """Remove an optimistic bubble that never became a real message.""" try: self._messages.remove(message) except ValueError: # pragma: no cover - a defensive check pass #: The blocks of Unicode where an emoji with the emoji presentation can be. A wide symbol #: between them is a CJK symbol or radical, not an emoji. _EMOJI_BLOCKS = ((0x2000, 0x2E7F), (0x1F000, 0x1FAFF)) #: The I Ching symbols, which Unicode 16 draws wide. They are symbols, not emoji: the #: firmware and the other apps never offer them. _NOT_EMOJI = frozenset(range(0x268A, 0x2690)) | frozenset(range(0x2630, 0x2638)) @lru_cache(maxsize=1) def emoji_catalog() -> tuple[str, ...]: """Each single emoji that the reaction dialog offers, in the order of Unicode. The standard library has no list of emoji. Thus the catalog takes two sets. The first is each symbol in the emoji blocks that Unicode draws wide by default (``🔥``, ``⌚``). The second is the emoji that are text by default (``❤``), which the table of Rich lists. Each of these gets the variation selector that asks for the emoji form (``❤️``), as the apps send it. A flag, a skin tone, and a joined sequence are not in the catalog, but a reaction with one still shows when it arrives. The names come from the Unicode database of the Python that runs MeshTerm, so a newer Python offers newer emoji. """ from rich._unicode_data import load found: dict[int, str] = {} for first, last in _EMOJI_BLOCKS: for cp in range(first, last + 1): char = chr(cp) if cp in _NOT_EMOJI: continue if unicodedata.category(char) == "So" and unicodedata.east_asian_width(char) == "W": found[cp] = char for char in load("auto").narrow_to_wide: if not char.isascii() and len(char) == 1: found.setdefault(ord(char), char + "\ufe0f") return tuple( emoji for _, emoji in sorted(found.items()) if not reactions._is_regional(emoji[0]) and reactions.is_reaction_emoji(emoji) ) async def our_node(ctx: AppContext) -> tuple[str | None, str | None]: """The name and the public key of our node, from the cache of the session. A chat shows them on our messages (refer to :func:`~meshterm.ui.widgets.self_chip`), and a direct chat names one end of each of its packets with the key. The read uses the cache, so it does not ask the device again. Returns: ``(name, key)``. An item is ``None`` when the device did not report it, or when the read failed. Our node with no name only keeps the bare ``★`` in a chat. """ try: info = await ctx.devstate.self_info() except Exception: # noqa: BLE001 - our node with no name keeps the bare star return None, None return str(info.get("name") or "") or None, str(info.get("public_key") or "") or None async def open_chat(ctx: AppContext, conversation: Conversation) -> int: """Open the live chat screen for ``conversation`` and run it until the user closes it. This function makes sure that the listener and the message recorder run, reads the stored transcript, subscribes to the hub so that inbound messages for this thread are appended live, and pushes the screen. At exit, it always removes the subscription and the marker of the active conversation. Args: ctx: The shared application context (it must run the interactive TUI surface). conversation: The channel or contact conversation to open. Returns: The number of messages in the transcript when the screen closed. Raises: RuntimeError: If the call is not from the interactive menu (no full-screen session). """ from .surface import TuiUi if not isinstance(ctx.ui, TuiUi): # pragma: no cover - the caller is only in the menu raise RuntimeError("live chat is only available in the interactive menu") if conversation.is_room: from .room import open_room return await open_room(ctx, conversation) session = ctx.ui.session device = await ctx.device() try: await ctx.chat.start() # start to store inbound messages, if it did not already except Exception: # noqa: BLE001 - the hub may already run. The recorder is best-effort pass from ..services import trace_runner history = ctx.repo.recent_chat_messages( is_channel=conversation.is_channel, channel_id=conversation.channel_id, peer=conversation.peer, limit=ctx.preferences.chat_history_limit, ) contacts = await ctx.devstate.contacts() names = _contact_names(contacts) stored_names, named_at = ctx.repo.node_names_with_times() key_of = trace_runner.make_name_key_resolver(contacts, stored_names, heard=named_at) async def send(text: str) -> ChatMessage | None: 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 _with_restore(ctx, lambda: ctx.chat.send_direct(conversation.contact, text)) resend: Callable[[ChatMessage], Awaitable[ChatMessage]] | None = None resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None scope: str | None = None if conversation.is_channel: scope = ctx.chat.channel_scope(conversation.channel_id) async def resend_unscoped(message: ChatMessage) -> ChatMessage | None: assert conversation.channel_idx is not None return await ctx.chat.send_channel( conversation.channel_idx, message.text, label=conversation.label, scope=WILDCARD ) else: async def resend(message: ChatMessage) -> ChatMessage: assert conversation.contact is not None return await _with_restore( ctx, lambda: ctx.chat.resend_direct(conversation.contact, message) ) from .map_screen import Locator from .node_detail_screen import ContactPages paths = await _make_paths_presenter(ctx, conversation, device) self_name, self_key = await our_node(ctx) screen = ChatScreen( conversation, history, send=send, names=names, session=session, resend=resend, paths=paths, key_of=key_of, scope=scope, resend_unscoped=resend_unscoped, locator=await Locator.gather(ctx), pages=await ContactPages.gather(ctx, self_key=self_key or ""), self_name=self_name, self_key=self_key, reaction_set=reactions.reaction_set(ctx.preferences.reaction_emoji), max_parts=ctx.preferences.message_parts_max, part_gap_s=ctx.preferences.trace_cooldown_s, ) 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, channel_id=conversation.channel_id ) ) 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) async def _with_restore(ctx: AppContext, attempt: Callable[[], Awaitable[Any]]) -> Any: """Run a direct send. If the device forgot the contact, offer to restore it, then retry. This is the one rejection from which a send can recover: the firmware has no contact for the recipient (refer to :class:`~meshterm.core.connection.ContactNotOnDeviceError`), and one write corrects it. The user is asked first (:func:`_restore_contact`). If the user declines, the error is raised again. Thus the chat shows the refusal exactly as it shows each other failed send. Args: ctx: The shared application context. attempt: The send to run. It is called a second time, with no change, when the contact is back. Returns: The return value of ``attempt``. Raises: Exception: Each error that ``attempt`` raises. A rejection for a forgotten contact is raised again only when the user declined the offer to add it back. """ try: return await attempt() except ContactNotOnDeviceError as missing: if not await _restore_contact(ctx, missing): raise return await attempt() async def _restore_contact(ctx: AppContext, missing: ContactNotOnDeviceError) -> bool: """Offer to write a forgotten contact back to the device. ``True`` if it now has it. A direct message is addressed through the contact entry of the *device*. Thus MeshTerm cannot send a message to a contact that the firmware removed, also when MeshTerm still lists it. (The list on a screen is the union of the table of the device and the contacts that MeshTerm keeps for it: refer to :mod:`meshterm.core.contact_store`.) The contact that we already have holds all the data that the entry must have. Thus one write is the repair, and the send that got the rejection tries again after it. The function offers the write and does not do it silently. The write changes what the device stores, and a full contact table refuses it (the user must see that, and it must not be hidden). Args: ctx: The shared application context. missing: The rejection, with the contact that the device could not find. Returns: ``True`` when the contact is on the device (retry the send). ``False`` if the user declined. Then the caller raises the error again, so that the chat shows the refusal. Raises: DeviceCommandError: If the device refused the write (most often, a full table). """ contact = missing.contact add = await ctx.ui.dialog( f"{contact.name} isn't in this device's contacts, so the radio can't address it. " "Add it back and send?", [("Cancel", False), ("Add & send", True)], title="Contact not on device", default=1, ) if not add: return False device = await ctx.device() await device.add_contact(contact) # The table of the device changed under the session cache. The next read gets it again # (with the route that the firmware learns for the contact from now on). ctx.devstate.invalidate_contacts() return True def _contact_names(contacts: list[Contact]) -> dict[str, str]: """Build a map from key prefix to name, for the labels of 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 that the screen shows now. Returns: ``True`` if the message must be appended 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) # --- message paths (the ^P view) ----------------------------------------------------- def _sent_scope_line(ctx: AppContext, message: ChatMessage, *, relayed: bool) -> Text | None: """The scope of a message that we sent, for the paths dialog, and why it can be lost. The line comes from the scope stored on the message at send time (:attr:`ChatMessage.scope`), never from the current scope of the channel, which may have changed since then. A scoped message can have no stored relayed copy *and* no repeater that was ever heard here to relay its region (:meth:`~meshterm.core.region_store.RegionStore.carriers`). Then the line gives the most probable reason in plain words: only repeaters that relay its region relay a scoped flood, and no known repeater in range does. Args: ctx: The shared application context (for the region store). message: The message whose paths are open. relayed: Whether a copy of it was heard when it came back. Returns: The line, or ``None`` for a message with no stored scope (inbound, direct, or sent before MeshTerm stored scopes). """ from ..core.regions import UNSCOPED, Scope from .widgets import scope_text if not message.outbound or not message.scope: return None line = Text("sent ", style="muted") if message.scope == WILDCARD: line.append_text(scope_text(UNSCOPED)) return line line.append("under ", style="muted") line.append_text(scope_text(Scope("scoped", message.scope))) store = getattr(ctx, "region_store", None) carriers = len(store.carriers(message.scope)) if store is not None else 0 if not relayed and not carriers: line.append(" — no repeater known here carries it", style="warn") elif carriers: line.append( f" · {carriers} known repeater{'s' if carriers != 1 else ''} carr" f"{'y' if carriers != 1 else 'ies'} it", style="muted", ) return line async def _make_paths_presenter( ctx: AppContext, conversation: Conversation, device, # noqa: ANN001 - the core Device. Its type is set at the source ) -> Callable[[ChatMessage], Awaitable[None]]: """Build the async presenter for the ^P delivery-paths view of the chat. It collects what the presenter must have, one time each time the chat opens: * a resolver of hop names over the contacts and all the names that the recorder ever heard (the rule of the app is that a node that has a name never shows as a bare hash), * the name of our node, for the white ``you``, * the width of the routing prefix, for the lit hashes, and * for a channel, its secret. The function reads it from the device when the conversation did not have one (a conversation from the picker knows its identity, but not always its key). Args: ctx: The shared application context. conversation: The conversation that the chat screen opens. device: The connected device (the caller already awaited it). Returns: An async callable that shows the paths of one message in a floating window. """ from ..services import trace_runner from ..services.message_paths import ( channel_arrivals, collapse, direct_arrivals, distinct_paths, message_scope, ) from .message_paths_screen import MessagePathsScreen session = ctx.ui.session contacts = await ctx.devstate.contacts() stored_names, named_at = ctx.repo.node_names_with_times() resolve = trace_runner.make_node_resolver(contacts, stored_names) type_of = trace_runner.make_node_type_resolver(contacts) key_of = trace_runner.make_name_key_resolver(contacts, stored_names, heard=named_at) prefix_bytes = await ctx.devstate.routing_prefix_bytes() # Our key names one end of each direct packet in this conversation. The key of the # contact names the other end (refer to :func:`direct_arrivals`). Our node with no name # only gets no white highlight. self_name, self_key = await our_node(ctx) secret = conversation.secret if conversation.is_channel and not secret and conversation.channel_idx is not None: try: payload = await device.get_channel(conversation.channel_idx) raw = (payload or {}).get("channel_secret") secret = bytes(raw) if raw else None except Exception: # noqa: BLE001 - no key, no decrypt. The view says so correctly secret = None async def present(message: ChatMessage) -> None: if conversation.is_channel and secret: arrivals = channel_arrivals( ctx.repo, message, channel_name=conversation.label, secret=secret ) matched = True summary = ( f"heard {len(arrivals)} time{'s' if len(arrivals) != 1 else ''}" f" · {distinct_paths(arrivals)} distinct " f"path{'s' if distinct_paths(arrivals) != 1 else ''}" if arrivals else "no copies in the packet log" ) elif conversation.is_channel: await session.scroll( Text( "This channel's key isn't at hand, so overheard frames can't be " "matched to the message.", style="muted", ), title="Message paths", ) return else: contact = conversation.contact peer_key = conversation.peer or ( (contact.public_key or contact.key_prefix) if contact else None ) arrivals, exact = direct_arrivals( ctx.repo, message, self_key=self_key, peer_key=peer_key ) # An exact match says the same as the channel view: these packets are this # message. The shared MAC of the packets gives the match. Thus it is correct, # also when MeshTerm cannot read one word of them. matched = exact arrivals = collapse(arrivals) heard = sum(a.copies for a in arrivals) if exact: summary = ( f"heard {heard} time{'s' if heard != 1 else ''}" f" · {len(arrivals)} distinct path{'s' if len(arrivals) != 1 else ''}" if arrivals else "no copies in the packet log" ) else: summary = ( "direct frames are encrypted — matched by address and time" if peer_key and self_key else "direct frames are encrypted — matched by time alone" ) # The two ends of the path: the node that the message started from, and the node # that it was addressed to. Our sends go from our node to the peer. An inbound # channel message names its sender on the air and is addressed to all nodes, so it # ends on our node. The origin of an inbound message in a direct chat is the peer of # the conversation. destination: str | None = self_name if message.outbound: source = self_name # A channel broadcast is addressed to no specific node, and the copies that we # log are its rebroadcasts that come back. Thus it really ends on our node. A # direct send does not: it went to one node, and its route ends there. if not conversation.is_channel: destination = conversation.label elif conversation.is_channel: source, _body = split_channel_sender(message.text) else: source = conversation.label await session.run_screen( MessagePathsScreen( message, arrivals, matched=matched, resolve=resolve, prefix_bytes=prefix_bytes, self_name=self_name, summary=summary, source=source or None, destination=destination or None, type_of=type_of, key_of=key_of, scope=message_scope(arrivals, ctx.region_store.scope_of), sent_scope=_sent_scope_line(ctx, message, relayed=bool(arrivals)), ) ) return present