Send a long chat message as numbered parts

A message that is too long for one packet now goes out as up to
message_parts_max parts (default 3, 1 turns it off), each ending with
its part number: [1/3], [2/3], [3/3]. Each part, with its part number,
fits in the byte limit of the conversation (130 for a channel, 150 for
a direct message). A cut goes at a space, and a word with no space to
cut at is cut between characters, never inside an emoji sequence or
after a combining accent.

Channel parts wait the transmit cooldown between them, so that the
floods do not go out as a burst. A direct part waits for the
acknowledgement of the part before it, or the end of its wait. An
unacknowledged part still lets the rest go out, and ^R retries each
one. A hard failure stops the rest and puts their text back in the
compose line. The byte gauge shows the number of parts and the bytes
of the last part. Rooms are not changed.

Co-Authored-By: Claude <noreply@anthropic.com>

docs/cli/README.md

@@ -1734,6 +1734,7 @@ flood_advert_cooldown_s      60                                    60
 weekly_flood_advert          off                                   off                                   Flood an advert after a week without one
 advert_quiet_s               30                                    30                                    Silence to wait for before the weekly advert
 direct_message_soft_retries  0                                     0                                     Times to resend a message that gets no reply
+message_parts_max            3                                     3                                     Send a long message as this many parts at most
 tx_opt_min                   18                                    18                                    Weakest power it will try
 tx_opt_max                   28                                    28                                    Strongest power it will try
 tx_snr_tolerance_db          1                                     1                                     Signals this close are a tie, so less power wins

docs/development/writing-style.md

@@ -194,6 +194,8 @@ and in quoted text.
 | packet | One radio transmission | frame (for radio data) |
 | protocol frame | One unit of the companion protocol on USB, Bluetooth, or TCP: a command, a reply, or a push from the companion. In code that handles only this protocol, "frame" alone is permitted | packet (for this) |
 | message | A text message, direct or on a channel | |
+| part | One of the messages that MeshTerm sends for a message that is too long for one packet | chunk, segment |
+| part number | The `[1/3]` at the end of a part | marker (for this) |
 | send | MeshTerm gives a message, a packet, or a command to the device | emit |
 | transmit | A radio puts a packet on the air | |
 | broadcast | Transmit to all nodes, with no destination | |

docs/guide/features.md

@@ -15,7 +15,7 @@ This page lists each interactive screen with its scripted equivalent, where one
 
 | Feature | What it does | Scripted |
 | --- | --- | --- |
-| **💬 Chat** | Live full-screen messages on channels and to contacts. The transcript scrolls, the input line stays at the bottom, and sent and received messages appear together. The rooms that you joined are also here. Each post shows under the name of its author. MeshTerm stores each message. Unread counts show in the menu header. | `meshterm chat send / history / list / listen` |
+| **💬 Chat** | Live full-screen messages on channels and to contacts. The transcript scrolls, the input line stays at the bottom, and sent and received messages appear together. The rooms that you joined are also here. Each post shows under the name of its author. A message that is too long for one packet goes out as parts, each ending with its number (`[1/3]`). MeshTerm stores each message. Unread counts show in the menu header. | `meshterm chat send / history / list / listen` |
 | **📻 Channels** | Create, join, reorder, mute, and share mesh channels. You can share a channel with a QR code or a `meshcore://` link. If you give a channel a **send scope**, its messages are flooded into one region. Only the repeaters that carry that region relay them. When you change to a new device, export all the channels to a file, then import the file on the new device. The **channel guide** lists the public channels that MeshTerm heard, also the channels that are not on your device, and adds one with Enter. | `meshterm channels list / add / join / import / share / clear / scope / export / guide` |
 | **📌 Rooms** | Each room server that your radio knows. A room server is a message board on a radio. It keeps the latest posts and sends you the posts that you missed. Join a room with its password. If a room does not answer, MeshTerm tells you what it knows and offers the next step to try. For example, it can send again by flood when the route that your radio learned is no longer good. | `meshterm rooms list / join / forget` |
 | **📨 Courier** | A store-and-forward outbox for contacts that you cannot reach yet. Queue a message. MeshTerm sends it when the contact is next heard, or at a time that you set. It tracks the acks, and it waits longer between each try (exponential backoff). | `meshterm courier queue / list / send / cancel / clear` |

meshterm/core/parts.py

@@ -0,0 +1,176 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Cut a message that is too long for one packet into parts that each fit.
+
+A MeshCore message must fit in one LoRa packet (refer to
+:data:`~meshterm.ui.tui.prompt.DM_BYTE_LIMIT`). When the text of a message is longer than
+this limit, MeshTerm can send it as more than one message. Each of these messages is a
+**part**, and each part ends with its **part number**: ``[1/3]``, ``[2/3]``, ``[3/3]``. The
+part number goes at the end, so that the start of the first part (the ``@[mention]`` of a
+reply) stays where a person reads first. Each part, with its part number, fits in the
+limit.
+
+A cut goes at a space when the window of the part has one. Thus each word stays complete.
+A word that is longer than a part (a long URL) has no space to cut at, and MeshTerm cuts it
+between two characters. Such a cut never divides a character, and it does not go into a
+sequence that shows as one symbol: an accent after its letter, a skin tone or a variation
+selector after its emoji, or the two sides of a zero-width joiner.
+
+The receiver needs nothing to read the parts. Each part is an ordinary message.
+"""
+
+from __future__ import annotations
+
+import re
+import unicodedata
+
+#: The marks that join to the character before them. A cut before one of these divides a
+#: symbol that the user sees as one: U+200D zero-width joiner, U+FE0E and U+FE0F variation
+#: selectors, U+20E3 combining keycap.
+_JOINERS = frozenset("‍︎️⃣")
+
+_SPACE = re.compile(r"\s")
+
+
+def _bytes(text: str) -> int:
+    """The UTF-8 length of ``text``: what counts against the limit of a packet."""
+    return len(text.encode("utf-8"))
+
+
+def part_number(index: int, total: int) -> str:
+    """The text that ends part ``index`` of ``total``: a space, then ``[index/total]``."""
+    return f" [{index}/{total}]"
+
+
+def _joins_previous(ch: str) -> bool:
+    """Whether ``ch`` joins to the character before it, so that a cut cannot go before it."""
+    code = ord(ch)
+    return (
+        ch in _JOINERS
+        or unicodedata.category(ch) in ("Mn", "Me", "Mc")
+        or 0x1F3FB <= code <= 0x1F3FF  # skin tone modifiers
+        or 0xE0020 <= code <= 0xE007F  # tag characters (subdivision flags)
+    )
+
+
+def _hard_cut(text: str, end: int) -> int:
+    """Move a cut at ``end`` back, so that it does not divide a symbol.
+
+    Returns the new cut. It is never less than 1, so that each part has text.
+    """
+    cut = end
+    while cut > 1 and (_joins_previous(text[cut]) or text[cut - 1] == "‍"):
+        cut -= 1
+    return cut if cut > 1 else end
+
+
+def _fit(text: str, start: int, budget: int) -> int:
+    """The end of the longest run of ``text`` from ``start`` that fits in ``budget`` bytes."""
+    used = 0
+    end = start
+    while end < len(text):
+        used += _bytes(text[end])
+        if used > budget:
+            break
+        end += 1
+    return end
+
+
+def _spans(text: str, budget: int) -> list[tuple[int, int]]:
+    """Cut ``text`` into runs of ``budget`` bytes or less, as ``(start, end)`` indexes.
+
+    The runs have no whitespace at their two ends. The whitespace between two runs is not
+    in a run.
+    """
+    spans: list[tuple[int, int]] = []
+    start = 0
+    while start < len(text) and text[start].isspace():
+        start += 1
+    while start < len(text):
+        end = _fit(text, start, budget)
+        if end == len(text):
+            spans.append((start, end))
+            break
+        if end == start:  # one character is larger than the budget: it cannot fit at all
+            end = start + 1
+        # Look for the last space in the window, and also at the first character after it:
+        # a space there means that the window ends at the end of a word.
+        space = next(
+            (i for i in range(end, start, -1) if _SPACE.match(text[i])),
+            None,
+        )
+        if space is not None:
+            stop = space
+            while stop > start and text[stop - 1].isspace():
+                stop -= 1
+            spans.append((start, stop))
+            start = space
+        else:
+            cut = start + _hard_cut(text[start:], end - start)
+            spans.append((start, cut))
+            start = cut
+        while start < len(text) and text[start].isspace():
+            start += 1
+    return spans
+
+
+def _plan(text: str, limit: int, max_parts: int) -> tuple[list[tuple[int, int]], int]:
+    """Cut ``text`` for ``max_parts`` parts at most, as spans and the total of the cut.
+
+    The size of a part number changes with the total (``[9/9]`` is shorter than
+    ``[10/10]``). Thus the budget for the text of a part comes from a total, and the cut
+    gives a new total. The plan starts at two parts and goes up until the two agree. The
+    returned spans can be more than ``max_parts``: the caller then knows where the text
+    that does not fit starts.
+    """
+    total = 2
+    while True:
+        budget = limit - _bytes(part_number(total, total))
+        spans = _spans(text, max(1, budget))
+        if len(spans) <= total or total >= max_parts:
+            return spans, total
+        total = min(max_parts, max(total + 1, len(spans)))
+
+
+def split(text: str, limit: int, max_parts: int) -> list[str] | None:
+    """Cut ``text`` into the parts that MeshTerm sends, each with its part number.
+
+    Args:
+        text: The text of the message.
+        limit: The UTF-8 byte limit of one message.
+        max_parts: The largest number of parts. With 1, MeshTerm never cuts a message.
+
+    Returns:
+        The texts to send, in order. A text that fits in ``limit`` is one part with no part
+        number. ``None`` when the text needs more than ``max_parts`` parts.
+    """
+    if _bytes(text) <= limit:
+        return [text]
+    if max_parts < 2:
+        return None
+    spans, _ = _plan(text, limit, max_parts)
+    if len(spans) > max_parts:
+        return None
+    total = len(spans)
+    return [text[a:b] + part_number(i, total) for i, (a, b) in enumerate(spans, 1)]
+
+
+def overflow_at(text: str, limit: int, max_parts: int) -> int | None:
+    """The index of the first character of ``text`` that does not fit, or ``None``.
+
+    With ``max_parts`` of 1, this is the first character whose bytes go past ``limit``.
+    With more, it is the start of the text that would go in the part after the last one.
+    """
+    if _bytes(text) <= limit:
+        return None
+    if max_parts < 2:
+        return _fit(text, 0, limit)
+    spans, _ = _plan(text, limit, max_parts)
+    if len(spans) <= max_parts:
+        return None
+    return spans[max_parts][0]
+
+
+def unnumbered(part: str, index: int, total: int) -> str:
+    """The text of ``part`` without its part number (``part`` itself when it has none)."""
+    tail = part_number(index, total)
+    return part[: -len(tail)] if total > 1 and part.endswith(tail) else part

meshterm/core/preferences.py

@@ -290,6 +290,19 @@ PREFERENCES: tuple[PrefSpec, ...] = (
         minimum=0,
         maximum=2,
     ),
+    PrefSpec(
+        key="message_parts_max",
+        label="Message parts",
+        help="Send a long message as this many parts at most",
+        group="Sending",
+        value_type="int",
+        # Each part is a transmission, and a channel part is a flood. Three parts hold
+        # about 370 bytes on a channel. With 1, MeshTerm never cuts a message, and a long
+        # message is refused as before. Refer to meshterm.core.parts.
+        default=3,
+        minimum=1,
+        maximum=9,
+    ),
     PrefSpec(
         key="room_relogin_minutes",
         label="Room re-login",

meshterm/ui/chat.py

@@ -25,7 +25,7 @@ from rich.cells import cell_len
 from rich.console import Group, RenderableType
 from rich.text import Text
 
-from ..core import links
+from ..core import links, parts
 from ..core.channels import MENTION, split_channel_sender
 from ..core.connection import ContactNotOnDeviceError
 from ..core.events import EventKind, MeshEvent
@@ -222,6 +222,8 @@ class ChatScreen(Screen):
         resend_unscoped: Callable[[ChatMessage], Awaitable[ChatMessage | None]] | None = None,
         locator: Locator | None = None,
         pages: ContactPages | None = None,
+        max_parts: int = 1,
+        part_gap_s: float = 0.0,
     ) -> None:
         """Build the chat screen.
 
@@ -259,6 +261,12 @@ class ChatScreen(Screen):
             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.
+            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.
         """
         super().__init__()
         self._scope = scope if conversation.is_channel else None
@@ -285,6 +293,8 @@ class ChatScreen(Screen):
         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
@@ -498,22 +508,33 @@ class ChatScreen(Screen):
         return len(self._editor.text.encode("utf-8"))
 
     def _overflow_at(self, limit: int) -> int | None:
-        """The index of the first compose character with bytes past ``limit``, or ``None``.
+        """The index of the first compose character that cannot be sent, or ``None``.
 
-        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.
+        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.
         """
-        total = 0
-        for i, ch in enumerate(self._editor.text):
-            total += len(ch.encode("utf-8"))
-            if total > limit:
-                return i
-        return None
+        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`)."""
-        return byte_counter(self._used_bytes(), limit)
+        """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)."""
@@ -1189,17 +1210,22 @@ class ChatScreen(Screen):
     def _submit(self) -> None:
         """Send the current input line as a message (scheduled from the key handler).
 
-        A line over the limit is refused. The buffer stays the same (so that the user can
-        make it shorter), and the status shows the overage, the same as the red overflow
-        that the compose bar already shows.
+        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
-        limit = self._byte_limit()
-        over = self._used_bytes() - limit
-        if over > 0:
-            self._status = f"Too long by {over} byte{'s' if over != 1 else ''} — trim to send."
+        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 = LineEditor()
@@ -1208,26 +1234,53 @@ class ChatScreen(Screen):
         if self._is_channel:
             self._status = "sending…"
             self._session.invalidate()
-            self._spawn(self._send_channel(text))
+            self._spawn(self._send_channel(cut))
         else:
-            # A direct chat shows an optimistic bubble. The mark at its end shows the
-            # delivery: a spinner now, then ✓/✗ when the ack resolves (or times out).
-            # acked=None ⇒ the spinner still turns.
-            pending = ChatMessage(text=text, outbound=True, created_at=utcnow())
-            self._messages.append(pending)
             self._status = ""
             self._session.invalidate()
-            self._spawn(self._send_direct(pending))
+            self._spawn(self._send_direct(cut))
 
-    async def _send_channel(self, text: str) -> None:
-        """Broadcast a channel message, and append it when the companion accepts it."""
+    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 = LineEditor(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:
-            message = await self._send(text)
-            if message is not None:
-                self._messages.append(message)
+            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._messages.append(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
@@ -1264,25 +1317,40 @@ class ChatScreen(Screen):
             except Exception:  # noqa: BLE001 - a spinner problem must never break a send
                 pass
 
-    async def _send_direct(self, pending: ChatMessage) -> None:
-        """Await the delivery of the optimistic ``pending`` bubble, then put in the stored row.
+    async def _send_direct(self, cut: list[str]) -> None:
+        """Send each part of a direct message, and show each as an optimistic bubble.
 
-        On success, the stored message replaces the pending bubble. The stored message has
-        its resolved ``acked`` state and its row id, so that the user can retry a ✗ later
-        in place. A hard failure (the companion refuses the send) removes the bubble and
-        shows the error inline, the same as a failed send always did.
+        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:
-            message = await self._spin_while(self._send(pending.text))
-        except Exception as exc:  # noqa: BLE001 - show the error inline, keep the chat open
-            self._discard(pending)
-            self._status = f"send failed: {exc}"
-        else:
-            if message is not None:
+            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)
-            else:
-                self._discard(pending)
             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
@@ -1517,6 +1585,8 @@ async def open_chat(ctx: AppContext, conversation: Conversation) -> int:
         resend_unscoped=resend_unscoped,
         locator=await Locator.gather(ctx),
         pages=await ContactPages.gather(ctx),
+        max_parts=ctx.preferences.message_parts_max,
+        part_gap_s=ctx.preferences.trace_cooldown_s,
     )
     ctx.chat.set_active(conversation.key)
 

tests/test_parts.py

@@ -0,0 +1,249 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for the parts of a long message: the cut, and how the chat screen sends them."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from meshterm.core import parts
+from meshterm.core.models import ChatMessage, Contact, Conversation
+from meshterm.core.preferences import Preferences
+from meshterm.ui.chat import ChatScreen
+
+LIMIT = 130
+
+_TEXT = " ".join(f"word{i:03d}" for i in range(40))  # 319 bytes, words of 7 characters
+
+
+def _bodies(cut: list[str]) -> list[str]:
+    return [parts.unnumbered(p, i, len(cut)) for i, p in enumerate(cut, 1)]
+
+
+# --- the cut -----------------------------------------------------------------------------
+
+
+def test_a_text_that_fits_is_one_part_with_no_part_number() -> None:
+    """A text in the limit goes out as it is, with no part number."""
+    assert parts.split("hello", LIMIT, 3) == ["hello"]
+    assert parts.split("x" * LIMIT, LIMIT, 3) == ["x" * LIMIT]
+
+
+def test_each_part_fits_and_ends_with_its_part_number() -> None:
+    """Each part, with its part number, fits in the limit."""
+    cut = parts.split(_TEXT, LIMIT, 3)
+    assert cut is not None and len(cut) == 3
+    for i, part in enumerate(cut, 1):
+        assert len(part.encode("utf-8")) <= LIMIT
+        assert part.endswith(f" [{i}/3]")
+
+
+def test_the_parts_cut_at_spaces_and_keep_all_the_words() -> None:
+    """Each cut is at a space, and the parts together hold all the words."""
+    cut = parts.split(_TEXT, LIMIT, 3)
+    assert cut is not None
+    bodies = _bodies(cut)
+    assert " ".join(bodies) == _TEXT
+    for body in bodies:
+        assert body == body.strip()
+        assert all(word.startswith("word") and len(word) == 7 for word in body.split(" "))
+
+
+def test_a_word_longer_than_a_part_is_cut_between_characters() -> None:
+    """A word with no space to cut at is cut between two characters."""
+    url = "https://example.com/" + "a" * 200
+    cut = parts.split(url, LIMIT, 3)
+    assert cut is not None and len(cut) == 2
+    assert "".join(_bodies(cut)) == url
+    assert all(len(p.encode("utf-8")) <= LIMIT for p in cut)
+
+
+def test_a_cut_never_divides_an_emoji_sequence_or_an_accent() -> None:
+    """A cut does not divide a joined emoji, or an accent from its letter."""
+    family = "👨‍👩‍👧"  # three emoji and two zero-width joiners: one symbol
+    cut = parts.split(family * 20, LIMIT, 9)
+    assert cut is not None
+    for body in _bodies(cut):
+        assert body == family * (len(body) // len(family))
+    accented = "é" * 100  # e and a combining acute accent, 3 bytes for each pair
+    cut = parts.split(accented, LIMIT, 9)
+    assert cut is not None
+    for body in _bodies(cut):
+        assert not body.startswith("́")
+        assert body.endswith("́")
+
+
+def test_a_text_that_needs_too_many_parts_is_refused() -> None:
+    """A text that needs more than the largest number of parts has no cut."""
+    assert parts.split("x" * 400, LIMIT, 3) is None
+    assert parts.split("x" * (LIMIT + 1), LIMIT, 1) is None
+
+
+def test_the_part_number_grows_with_the_total() -> None:
+    """The part number shows the index and the total."""
+    assert parts.part_number(1, 3) == " [1/3]"
+    assert parts.part_number(2, 9) == " [2/9]"
+
+
+def test_overflow_at_finds_the_text_after_the_last_part() -> None:
+    """The overflow starts where the text after the last part starts."""
+    assert parts.overflow_at("x" * LIMIT, LIMIT, 1) is None
+    assert parts.overflow_at("x" * (LIMIT + 5), LIMIT, 1) == LIMIT  # the old rule
+    assert parts.overflow_at(_TEXT, LIMIT, 3) is None
+    long = "y" * 500
+    cut_at = parts.overflow_at(long, LIMIT, 3)
+    budget = LIMIT - len(parts.part_number(3, 3))
+    assert cut_at == 3 * budget
+
+
+def test_unnumbered_removes_only_its_own_part_number() -> None:
+    """Only the part number of the part itself is removed."""
+    assert parts.unnumbered("abc [2/3]", 2, 3) == "abc"
+    assert parts.unnumbered("abc [2/3]", 1, 3) == "abc [2/3]"
+    assert parts.unnumbered("abc", 1, 1) == "abc"
+
+
+def test_the_preference_permits_three_parts_by_default() -> None:
+    """The default of the preference is three parts."""
+    assert Preferences().message_parts_max == 3
+
+
+# --- the chat screen ---------------------------------------------------------------------
+
+
+class _Session:
+    def invalidate(self) -> None:
+        pass
+
+    def run_detached(self, work):  # noqa: ANN001, ANN201
+        return asyncio.ensure_future(work)
+
+
+def _screen(send, *, channel: bool, max_parts: int = 3, gap: float = 0.0) -> ChatScreen:
+    if channel:
+        conv = Conversation(label="#public", is_channel=True, channel_idx=0)
+    else:
+        conv = Conversation(
+            label="Alice",
+            is_channel=False,
+            contact=Contact(name="Alice", public_key="d4" + "0" * 62, key_prefix="d4e5f6a7"),
+        )
+    return ChatScreen(
+        conv,
+        [],
+        send=send,
+        names={},
+        session=_Session(),
+        max_parts=max_parts,
+        part_gap_s=gap,
+    )
+
+
+async def _enter(screen: ChatScreen, text: str) -> None:
+    screen.handle("text", text)
+    screen.handle("enter")
+    await asyncio.gather(*list(screen._tasks))
+
+
+async def test_a_channel_sends_each_part_in_order_with_the_gap(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A channel sends the parts in order, with the gap between them."""
+    sent: list[str] = []
+    waits: list[float] = []
+    real_sleep = asyncio.sleep
+
+    async def sleep(delay: float) -> None:
+        waits.append(delay)
+        await real_sleep(0)
+
+    async def send(text: str) -> ChatMessage:
+        sent.append(text)
+        return ChatMessage(text=text, outbound=True, is_channel=True, channel_idx=0)
+
+    screen = _screen(send, channel=True, gap=5.0)
+    monkeypatch.setattr("meshterm.ui.chat.asyncio.sleep", sleep)
+    await _enter(screen, _TEXT)
+
+    assert sent == parts.split(_TEXT, LIMIT, 3)
+    assert waits == [5.0, 5.0]  # between the parts, not before the first
+    assert [m.text for m in screen._messages] == sent
+    assert screen._status == ""
+    assert not screen._sending
+
+
+async def test_a_direct_message_sends_each_part_after_the_one_before() -> None:
+    """A direct part goes out only after the part before it."""
+    sent: list[str] = []
+    in_flight = 0
+
+    async def send(text: str) -> ChatMessage:
+        nonlocal in_flight
+        in_flight += 1
+        assert in_flight == 1  # one part at a time
+        await asyncio.sleep(0)
+        sent.append(text)
+        in_flight -= 1
+        return ChatMessage(text=text, outbound=True, peer="d4e5f6a7", acked=False)
+
+    screen = _screen(send, channel=False, max_parts=4)
+    await _enter(screen, _TEXT + " more")
+    assert len(sent) == 3
+    assert [m.text for m in screen._messages] == sent
+    assert all(m.acked is False for m in screen._messages)  # an unacked part did not stop the rest
+
+
+async def test_a_failed_part_stops_the_rest_and_keeps_their_text() -> None:
+    """A part that fails stops the parts after it, and their text goes back."""
+    calls = 0
+
+    async def send(text: str) -> ChatMessage:
+        nonlocal calls
+        calls += 1
+        if calls == 2:
+            raise RuntimeError("radio busy")
+        return ChatMessage(text=text, outbound=True, peer="d4e5f6a7", acked=True)
+
+    screen = _screen(send, channel=False)
+    await _enter(screen, _TEXT)
+    cut = parts.split(_TEXT, 150, 3)  # the limit of a direct message
+    assert cut is not None
+    assert calls == 2
+    assert len(screen._messages) == 1  # the failed bubble is gone
+    assert "radio busy" in screen._status
+    assert screen._editor.text == " ".join(_bodies(cut)[1:])  # parts 2 and 3, unnumbered
+
+
+async def test_too_many_parts_is_refused_and_the_text_stays() -> None:
+    """A text with too many parts is refused, and the text stays."""
+    sent: list[str] = []
+
+    async def send(text: str) -> ChatMessage:
+        sent.append(text)
+        return ChatMessage(text=text, outbound=True)
+
+    screen = _screen(send, channel=True, max_parts=2)
+    await _enter(screen, _TEXT)
+    assert sent == []
+    assert screen._editor.text == _TEXT
+    assert screen._status == "Too long for 2 parts — trim to send."
+
+
+def test_the_gauge_counts_the_parts() -> None:
+    """The gauge shows the number of parts and the bytes of the last part."""
+    screen = _screen(None, channel=True)
+    screen.handle("text", _TEXT)
+    gauge = screen._byte_counter(LIMIT).plain
+    cut = parts.split(_TEXT, LIMIT, 3)
+    assert cut is not None
+    assert gauge == f"3 parts · {len(cut[-1].encode('utf-8'))}/{LIMIT}"
+    assert screen._overflow_at(LIMIT) is None  # nothing is marked red
+
+
+def test_one_part_keeps_the_old_gauge_and_the_old_overflow() -> None:
+    """With one part, the gauge and the overflow are as before."""
+    screen = _screen(None, channel=True, max_parts=1)
+    screen.handle("text", "x" * (LIMIT + 2))
+    assert screen._byte_counter(LIMIT).plain == f"{LIMIT + 2}/{LIMIT}"
+    assert screen._overflow_at(LIMIT) == LIMIT