Acks get the wait a flood needs, and a late one still counts

Only 24 of 144 direct messages and posts in the history were ever marked
delivered. Three faults, each fixed:

- The wait was the radio's own estimate times 1.2: about 5.7 s for a
  flood, while a flooded round trip from a few hops out takes ten (a
  room's answer from five hops landed at 10.1 s on hardware). It is now
  the login's patience, the estimate stretched a quarter plus a round
  trip sized to the route, about 15 s for a flood.
- An ack that landed after the wait was dropped. The radio pushes one
  whenever it arrives, for any of its last eight sends, but nothing was
  listening, so a delivered message stayed failed for good. The history
  shows ACK and PATH packets heard seconds after messages recorded as
  failed. The send now returns a Delivery carrying the ack code, and the
  chat service keeps the last sixteen messages still waiting. A late ack
  flips one to delivered, in history and in the message the chat screen
  holds, and every code a retried message went out under counts.
- The Courier trusted the failure and sent the message again: a
  duplicate for the recipient whenever the ack was only late. A late ack
  now marks the entry delivered.

The ack listener is also armed before the send rather than after it,
so a neighbour's ack quicker than the send's own return is not missed.
The debug log names how a message went out and the radio's own
round-trip time.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

meshterm/core/connection.py

@@ -39,6 +39,7 @@ from .models import (
     NODE_TYPE_ROOM,
     Ack,
     Contact,
+    Delivery,
     Hop,
     LoginResult,
     Message,
@@ -1386,16 +1387,17 @@ class Device(ABC):
     # -- messaging ---------------------------------------------------------------
 
     @abstractmethod
-    async def send_direct_message(self, contact: Contact, text: str) -> Ack | None:
-        """Send a direct text message to a contact.
+    async def send_direct_message(self, contact: Contact, text: str) -> Delivery:
+        """Send a direct text message to a contact, and wait a while for its ack.
 
         Args:
             contact: The recipient; its ``public_key`` addresses the message.
             text: The message body.
 
         Returns:
-            The delivery :class:`Ack` if one arrived before the send timed out, else
-            ``None`` (the message was handed to the radio but not yet acknowledged).
+            The :class:`~meshterm.core.models.Delivery`: the ack code the radio expects
+            back, and the ack itself if it arrived within the wait. One that comes later is
+            still pushed by the radio, as an ``ACK`` event carrying the same code.
 
         Raises:
             DeviceCommandError: If the companion rejected the send outright.
@@ -3807,35 +3809,86 @@ class MeshCoreDevice(Device):
 
     async def send_direct_message(  # noqa: D102 - inherited docstring
         self, contact: Contact, text: str
-    ) -> Ack | None:
+    ) -> Delivery:
         transmit_gate.mark()
         from meshcore import EventType
 
         mc = self._require()
         pub = self._node_pubkey(contact)
-        # Only the hand-over holds the transmit lock; the ack wait below is not a send.
-        async with self.transmitting():
-            result = await mc.commands.send_msg(pub, text)
-        if result is None or getattr(result, "is_error", lambda: False)():
-            # A recipient the firmware has no entry for is the one rejection with an obvious
-            # fix, so it gets its own class for the chat screen to offer that fix on; see
-            # :class:`ContactNotOnDeviceError` for how a listed contact can be missing here.
-            if error_code(result) == _ERR_NOT_FOUND:
-                raise ContactNotOnDeviceError(contact)
-            raise DeviceCommandError(f"couldn't send to {contact.name}: {reject_reason(result)}")
-        # The companion acknowledges the send immediately with an ``expected_ack`` code and
-        # a suggested wait; the recipient's delivery ACK arrives later carrying that code.
-        payload = getattr(result, "payload", {}) or {}
-        expected = payload.get("expected_ack")
-        expected_hex = expected.hex() if isinstance(expected, (bytes, bytearray)) else expected
-        if not expected_hex:
-            return None
-        suggested = payload.get("suggested_timeout")
-        timeout = suggested / 1000 * 1.2 if suggested else 8.0
-        ack = await mc.wait_for_event(
-            EventType.ACK, attribute_filters={"code": expected_hex}, timeout=timeout
+        loop = asyncio.get_running_loop()
+        # Listen before sending, and to every ack: the code to match is only known once the
+        # send returns, and an ack quicker than that return (a neighbour's) used to land
+        # before anything was listening for it. Kept by code until the code is known.
+        heard: dict[str, object] = {}
+        wanted: asyncio.Future = loop.create_future()
+        code: list[str] = []
+        flood = False
+        budget = 0.0
+
+        def on_ack(event) -> None:  # noqa: ANN001 - meshcore Event
+            got = str((getattr(event, "payload", None) or {}).get("code") or "").lower()
+            heard[got] = event
+            if code and got == code[0] and not wanted.done():
+                wanted.set_result(event)
+
+        subscription = mc.subscribe(EventType.ACK, on_ack)
+        started = loop.time()
+        try:
+            # Only the hand-over holds the transmit lock; the ack wait below is not a send.
+            async with self.transmitting():
+                result = await mc.commands.send_msg(pub, text)
+            if result is None or getattr(result, "is_error", lambda: False)():
+                # A recipient the firmware has no entry for is the one rejection with an
+                # obvious fix, so it gets its own class for the chat screen to offer that
+                # fix on; see :class:`ContactNotOnDeviceError` for how a listed contact can
+                # be missing here.
+                if error_code(result) == _ERR_NOT_FOUND:
+                    raise ContactNotOnDeviceError(contact)
+                raise DeviceCommandError(
+                    f"couldn't send to {contact.name}: {reject_reason(result)}"
+                )
+            # The companion accepts the send at once with the ``expected_ack`` code and how
+            # it sent it; the recipient's ack arrives later, carrying the code.
+            payload = getattr(result, "payload", {}) or {}
+            expected = payload.get("expected_ack")
+            expected_hex = expected.hex() if isinstance(expected, (bytes, bytearray)) else expected
+            if not expected_hex:
+                return Delivery(None)
+            code.append(str(expected_hex).lower())
+            if code[0] in heard:
+                wanted.set_result(heard[code[0]])
+            flood = payload.get("type") == 1
+            # As patient as a login, whose answer travels exactly as an ack does: the
+            # radio's own estimate (``suggested_timeout``, which the library stretches by a
+            # quarter), then a round trip sized to the route. The estimate alone (what this
+            # waited until 2026-10-05) is ``500 ms + 16 x airtime`` for a flood, about five
+            # seconds, where a flood's answer from five hops out took ten on hardware. So
+            # most acks from beyond a neighbour landed after the wait had given up, and four
+            # messages in five read as failed that had been delivered.
+            hops = 0 if flood else 2 * len(contact.route_hops or ())
+            suggested = float(payload.get("suggested_timeout") or 0) / 1000.0
+            budget = suggested * 1.25 + max(trace_timeout(hops), trace_timeout(0))
+            if not wanted.done():
+                try:
+                    await asyncio.wait_for(asyncio.shield(wanted), budget)
+                except asyncio.TimeoutError:
+                    pass
+        finally:
+            subscription.unsubscribe()
+        event = wanted.result() if wanted.done() else None
+        ack = ack_from_event(event) if event is not None else None
+        _log.debug(
+            "direct message to %s: sent by %s, code %s, budget=%.1fs -> %s",
+            contact.name,
+            "flood" if flood else "its route",
+            code[0],
+            budget,
+            f"acked in {(loop.time() - started) * 1000.0:.0f}ms "
+            f"(the radio's round trip: {(ack.raw or {}).get('trip_time')}ms)"
+            if ack is not None
+            else "no ack yet",
         )
-        return ack_from_event(ack) if ack is not None else None
+        return Delivery(code[0], ack, flood)
 
     async def send_channel_message(  # noqa: D102 - inherited docstring
         self, index: int, text: str
@@ -4313,6 +4366,12 @@ class MockDevice(Device):
         #: Whoever is subscribed to the event stream right now, so a room can push its posts
         #: to them after a login the way a real room's catch-up arrives — unsolicited.
         self._listeners: list[EventCallback] = []
+        #: Contacts whose acks arrive only after the send has stopped waiting, the common
+        #: case beyond a neighbour on a real mesh. Empty by default; name one to walk the
+        #: late-ack path.
+        self._late_acks: set[str] = set()
+        #: Direct messages handed over so far, which numbers each one's ack code.
+        self._sent_codes = 0
         #: Nodes whose learned route has gone stale: a message sent along it meets silence,
         #: and a flood gets through and learns a fresh one. Empty by default; name a contact
         #: here to walk the "try by flood" path a stale route leaves a reader on.
@@ -4489,17 +4548,38 @@ class MockDevice(Device):
         # chat screen's "add it back and send" offer walkable on the simulator.
         if self._mock_key(contact) not in {self._mock_key(c) for c in self._contacts}:
             raise ContactNotOnDeviceError(contact)
+        self._sent_codes += 1
+        code = f"{self._sent_codes:08x}"
+        held = next(c for c in self._contacts if self._mock_key(c) == self._mock_key(contact))
+        flood = held.route_hops is None
         if contact.is_room:
             # A post. The room keeps it — and acknowledges — only from a member it knows
             # who may post; anyone else's is dropped with no reply, which is all a read-only
             # member ever learns about why a post went nowhere.
             access = self._room_access.get(self._mock_key(contact))
             if access is None or not access.can_post:
-                return None
+                return Delivery(code, None, flood)
             self._room_board.append((self._self_prefix(), utcnow(), text))
+        if contact.name in self._late_acks:
+            # Delivered, but the ack takes the long way home: it arrives after the wait,
+            # pushed the way the radio pushes one whenever it lands.
+            self._push_ack_later(code)
+            return Delivery(code, None, flood)
         # Otherwise the simulator "delivers" instantly and always acknowledges, so outbound
         # direct messages show as acked without a radio.
-        return Ack(code="mock")
+        return Delivery(code, Ack(code=code), flood)
+
+    def _push_ack_later(self, code: str) -> None:
+        """Push an ack to whoever listens, a beat after the send returned without one."""
+
+        async def push() -> None:
+            await asyncio.sleep(_MOCK_MONITOR_INTERVAL_S)
+            for listener in list(self._listeners):
+                listener(MeshEvent.ack_event(Ack(code=code)))
+
+        task = asyncio.create_task(push())
+        self._bg_tasks.add(task)
+        task.add_done_callback(self._bg_tasks.discard)
 
     async def send_channel_message(  # noqa: D102 - inherited docstring
         self, index: int, text: str

meshterm/core/models.py

@@ -472,6 +472,36 @@ class Ack:
     raw: dict | None = None
 
 
+@dataclass(slots=True)
+class Delivery:
+    """A direct message the radio took: the ack it is waiting for, and whether it came in time.
+
+    The ack is the recipient's proof of receipt (for a room, of *storing* the post). It is
+    keyed by a four-byte code the radio derives from the message and hands back the moment
+    it accepts the send; the ack, whenever it arrives, carries the same code. Keeping the
+    code is what lets an ack that lands after the wait still count — the radio pushes it
+    regardless, and a flood's ack, riding back on the path the message found, often takes
+    longer than any wait a person would sit through.
+
+    Attributes:
+        code: The ack code the radio expects back, lowercase hex; ``None`` when the radio
+            gave none (nothing to wait for, and nothing a late ack could match).
+        ack: The acknowledgement, when it arrived within the wait; ``None`` otherwise —
+            which is *not* "not delivered", only "not confirmed yet".
+        flood: How the radio sent it: ``True`` to the whole mesh, ``False`` along the route
+            it learned; ``None`` when it didn't say.
+    """
+
+    code: str | None
+    ack: Ack | None = None
+    flood: bool | None = None
+
+    @property
+    def acked(self) -> bool:
+        """Whether the recipient's acknowledgement arrived within the wait."""
+        return self.ack is not None
+
+
 def conversation_key(is_channel: bool, channel_id: str | None, peer: str | None) -> str:
     """Return a stable key identifying a chat conversation.
 

meshterm/services/chat_service.py

@@ -18,6 +18,8 @@ communications, not overheard noise, so they are always recorded while a device
 from __future__ import annotations
 
 import asyncio
+from collections import OrderedDict
+from collections.abc import Callable
 from typing import TYPE_CHECKING
 
 from ..core.channels import channel_identity
@@ -26,6 +28,7 @@ from ..core.events import EventKind, MeshEvent
 from ..core.models import (
     ChatMessage,
     Contact,
+    Delivery,
     Message,
     is_direct_messageable,
     utcnow,
@@ -36,6 +39,15 @@ if TYPE_CHECKING:
     from ..context import AppContext
 
 
+#: How many unacknowledged messages keep waiting for a late ack. The radio itself remembers
+#: the codes of only its last eight sends (MeshCore's ``EXPECTED_ACK_TABLE_SIZE``) and
+#: pushes an ack for none older, so a few more than that covers every ack still possible.
+_AWAITING_CAP = 16
+
+#: Called with a message once its ack arrives after the send stopped waiting for it.
+LateAck = Callable[[ChatMessage], None]
+
+
 def _fallback_channel_id(idx: int | None) -> str:
     """Identity for a channel we can't read (an unconfigured/unknown slot).
 
@@ -80,6 +92,11 @@ class ChatService:
         # slot fresh every time (see :meth:`channel_id_for`), so this is not the source of
         # truth — only a warm map for priming/backfill and a fallback when a read fails.
         self._channel_ids: dict[int, str] = {}
+        # Messages sent whose ack had not arrived when the send stopped waiting, by the ack
+        # code a late one would carry (every attempt's, for a retried message), oldest
+        # first, each with whoever wants to hear when it lands (see :meth:`_settle_ack`).
+        self._awaiting: OrderedDict[str, tuple[ChatMessage, LateAck | None]] = OrderedDict()
+        self._ack_unsubscribe: Unsubscribe | None = None
 
     @property
     def active(self) -> bool:
@@ -234,6 +251,7 @@ class ChatService:
                 queue.put_nowait(message)
 
         self._unsubscribe = self._ctx.events.subscribe(on_event, EventKind.MESSAGE)
+        self._ack_unsubscribe = self._ctx.events.subscribe(self._on_ack, EventKind.ACK)
         self._run_id = run_id
         self._ctx.log.info("chat recording (run %s)", run_id)
         await self._prime_channels()
@@ -264,8 +282,11 @@ class ChatService:
         try:
             assert self._unsubscribe is not None
             self._unsubscribe()
+            if self._ack_unsubscribe is not None:
+                self._ack_unsubscribe()
         finally:
             self._unsubscribe = None
+            self._ack_unsubscribe = None
         if self._worker is not None:
             self._worker.cancel()
             try:
@@ -434,7 +455,7 @@ class ChatService:
                     return is_direct_messageable(contact.node_type)
         return False
 
-    async def _deliver_direct(self, contact: Contact, text: str):
+    async def _deliver_direct(self, contact: Contact, text: str) -> tuple[bool, list[str]]:
         """Transmit a direct message, softly retrying until it is acknowledged.
 
         A single logical send makes one initial transmission plus up to
@@ -454,8 +475,8 @@ class ChatService:
             text: The message body.
 
         Returns:
-            The delivery :class:`~meshterm.core.models.Ack` from the first attempt that landed,
-            or ``None`` if every attempt went unacknowledged.
+            Whether an attempt was acknowledged within its wait, and the ack code of every
+            attempt made — any of which a late ack may still carry (the radio keeps them).
 
         Raises:
             Exception: Propagates a hard send failure (the companion rejecting the send); such
@@ -463,11 +484,13 @@ class ChatService:
         """
         device = await self._ctx.device()
         attempts = max(0, self._ctx.preferences.direct_message_soft_retries) + 1
-        ack = None
+        codes: list[str] = []
         for attempt in range(1, attempts + 1):
-            ack = await device.send_direct_message(contact, text)
-            if ack is not None:
-                return ack
+            delivery: Delivery = await device.send_direct_message(contact, text)
+            if delivery.code:
+                codes.append(delivery.code)
+            if delivery.acked:
+                return True, codes
             if attempt < attempts:
                 self._ctx.log.debug(
                     "chat: DM to %s unacked on try %d/%d; soft-retrying",
@@ -475,34 +498,41 @@ class ChatService:
                     attempt,
                     attempts,
                 )
-        return ack
+        return False, codes
 
-    async def send_direct(self, contact: Contact, text: str) -> ChatMessage:
+    async def send_direct(
+        self, contact: Contact, text: str, *, on_late_ack: LateAck | None = None
+    ) -> ChatMessage:
         """Send a direct message to a contact and record it in history.
 
         Delivery is attempted with up to ``preferences.direct_message_soft_retries`` automatic
         soft retries (see :meth:`_deliver_direct`) before the message is recorded as
-        unacknowledged.
+        unacknowledged — which means *not confirmed yet*: an ack that lands later still
+        settles it (see :meth:`_settle_ack`), in history and on the screen showing it.
 
         Args:
             contact: The recipient.
             text: The message body.
+            on_late_ack: Called with the recorded message if its ack arrives after the
+                send stopped waiting (the Courier, which would otherwise send it again).
 
         Returns:
             The recorded outbound :class:`ChatMessage` (its ``acked`` reflects whether a
-            delivery acknowledgement arrived within the soft-retry budget).
+            delivery acknowledgement arrived within the wait, so far).
         """
-        ack = await self._deliver_direct(contact, text)
+        acked, codes = await self._deliver_direct(contact, text)
         chat = ChatMessage(
             text=text,
             outbound=True,
             is_channel=False,
             peer=contact.key_prefix or contact.public_key[:12] or None,
             peer_name=contact.name,
-            acked=ack is not None,
+            acked=acked,
             created_at=utcnow(),
         )
         chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_id)
+        if not acked:
+            self._await_ack(chat, codes, on_late_ack)
         return chat
 
     async def resend_direct(self, contact: Contact, message: ChatMessage) -> ChatMessage:
@@ -523,10 +553,14 @@ class ChatService:
         Returns:
             The same ``message``, with :attr:`~ChatMessage.acked` refreshed.
         """
-        ack = await self._deliver_direct(contact, message.text)
-        message.acked = ack is not None
+        acked, codes = await self._deliver_direct(contact, message.text)
+        if message.acked:
+            return message  # a late ack for an earlier attempt settled it meanwhile
+        message.acked = acked
         if message.row_id is not None:
             self._ctx.repo.update_chat_ack(message.row_id, message.acked)
+        if not acked:
+            self._await_ack(message, codes, None)
         return message
 
     async def send_post(self, room: Contact, text: str) -> ChatMessage:
@@ -553,17 +587,19 @@ class ChatService:
             stored it.
         """
         device = await self._ctx.device()
-        ack = await device.send_direct_message(room, text)
+        delivery = await device.send_direct_message(room, text)
         chat = ChatMessage(
             text=text,
             outbound=True,
             is_channel=False,
             peer=room.key_prefix or room.public_key[:12] or None,
             peer_name=room.name,
-            acked=ack is not None,
+            acked=delivery.acked,
             created_at=utcnow(),
         )
         chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_id)
+        if not delivery.acked:
+            self._await_ack(chat, [delivery.code] if delivery.code else [], None)
         return chat
 
     async def resend_post(self, room: Contact, message: ChatMessage) -> ChatMessage:
@@ -577,12 +613,68 @@ class ChatService:
             The same ``message``, with :attr:`~ChatMessage.acked` refreshed.
         """
         device = await self._ctx.device()
-        ack = await device.send_direct_message(room, message.text)
-        message.acked = ack is not None
+        delivery = await device.send_direct_message(room, message.text)
+        if message.acked:
+            return message  # a late ack for the first post settled it meanwhile
+        message.acked = delivery.acked
         if message.row_id is not None:
             self._ctx.repo.update_chat_ack(message.row_id, message.acked)
+        if not delivery.acked:
+            self._await_ack(message, [delivery.code] if delivery.code else [], None)
         return message
 
+    # --- late acks ------------------------------------------------------------------------
+
+    def _await_ack(self, chat: ChatMessage, codes: list[str], on_late_ack: LateAck | None) -> None:
+        """Keep listening for a sent message's ack after the send has stopped waiting.
+
+        The radio pushes an ack whenever it lands, however late, for any of its last eight
+        sends; a flood's ack, riding home on the path the message found, routinely lands
+        after the wait. So the message is filed under every code it was sent with, the
+        oldest entries giving way past :data:`_AWAITING_CAP`.
+        """
+        for code in codes:
+            self._awaiting[code.lower()] = (chat, on_late_ack)
+            self._awaiting.move_to_end(code.lower())
+        while len(self._awaiting) > _AWAITING_CAP:
+            self._awaiting.popitem(last=False)
+
+    def _on_ack(self, event: MeshEvent) -> None:
+        """Hub callback: an ack arrived; settle the message it belongs to, if one waits."""
+        ack = event.ack
+        if ack is None or not ack.code:
+            return
+        self._settle_ack(str(ack.code).lower())
+
+    def _settle_ack(self, code: str) -> None:
+        """Mark the message waiting on ``code`` delivered, everywhere it is shown or kept.
+
+        The history row is updated, and the very :class:`ChatMessage` the chat screen
+        holds flips to acknowledged, so its ✗ turns to ✓ on the next paint. Whoever asked
+        to hear of it (the Courier) is told. Every other code the message was sent under
+        (an earlier attempt's) is let go: it was delivered once, and that is the news.
+        """
+        entry = self._awaiting.pop(code, None)
+        if entry is None:
+            return
+        chat, on_late_ack = entry
+        for other in [c for c, (held, _) in self._awaiting.items() if held is chat]:
+            del self._awaiting[other]
+        if chat.acked:
+            return
+        chat.acked = True
+        if chat.row_id is not None:
+            try:
+                self._ctx.repo.update_chat_ack(chat.row_id, True)
+            except Exception as exc:  # noqa: BLE001 - the screen still shows it delivered
+                self._ctx.log.debug("chat: failed to record a late ack: %s", exc)
+        self._ctx.log.info("chat: late ack from %s", chat.peer_name or chat.peer)
+        if on_late_ack is not None:
+            try:
+                on_late_ack(chat)
+            except Exception as exc:  # noqa: BLE001 - one listener must not break the hub
+                self._ctx.log.debug("chat: late-ack listener failed: %s", exc)
+
     def channel_scope(self, channel_id: str | None) -> str | None:
         """The region a channel's messages are sent under, or ``None`` for the device default.
 

meshterm/services/courier.py

@@ -30,7 +30,7 @@ from datetime import datetime
 from typing import TYPE_CHECKING
 
 from ..core.connection import Unsubscribe
-from ..core.courier_store import QUEUED, QueuedMessage
+from ..core.courier_store import DELIVERED, QUEUED, QueuedMessage
 from ..core.events import EventKind, MeshEvent
 from ..core.models import Contact, utcnow
 
@@ -218,7 +218,11 @@ class CourierService:
                 self._ctx.log.debug("courier: no contact for %r; leaving queued", message.node_name)
                 return "unknown contact"
             store.note_attempt(message.ident)
-            chat = await self._ctx.chat.send_direct(contact, message.text)
+            chat = await self._ctx.chat.send_direct(
+                contact,
+                message.text,
+                on_late_ack=lambda _chat, ident=message.ident: self._delivered_late(ident),
+            )
             if chat.acked:
                 store.mark_delivered(message.ident)
                 self._ctx.watch_store.add_alert(
@@ -246,6 +250,24 @@ class CourierService:
         finally:
             self._sending = False
 
+    def _delivered_late(self, ident: int) -> None:
+        """An attempt's ack arrived after the attempt stopped waiting: it was delivered.
+
+        Without this the entry stayed queued (or gave up) and the next attempt sent the
+        recipient the same message again — every late ack a duplicate on their screen.
+        """
+        store = self._ctx.courier_store
+        message = store.get(ident)
+        if message is None or message.status == DELIVERED:
+            return
+        store.mark_delivered(ident)
+        self._ctx.watch_store.add_alert(
+            "courier",
+            message.node_name,
+            f"delivered “{_preview(message.text)}” (attempt {message.attempts}, ack came late)",
+        )
+        self._ctx.log.info("courier: delivered to %s (late ack)", message.node_name)
+
     async def _resolve(self, message: QueuedMessage) -> Contact | None:
         """Find the recipient in the device's contact list, by key then by name."""
         device = await self._ctx.device()

tests/test_chat.py

@@ -113,8 +113,8 @@ async def test_mock_send_direct_returns_ack() -> None:
     device = MockDevice()
     await device.connect()
     contact = (await device.get_contacts())[0]
-    ack = await device.send_direct_message(contact, "hello")
-    assert ack is not None
+    delivery = await device.send_direct_message(contact, "hello")
+    assert delivery.acked and delivery.code
     await device.send_channel_message(0, "hi channel")  # no return, must not raise
     await device.disconnect()
 
@@ -146,7 +146,7 @@ async def test_adding_a_forgotten_contact_back_makes_it_messageable() -> None:
     stranger = Contact(name="bob_caribou", public_key="a3" * 32, key_prefix="a3a3a3a3")
     await device.add_contact(stranger)
     assert "bob_caribou" in {c.name for c in await device.get_contacts()}
-    assert await device.send_direct_message(stranger, "hello") is not None
+    assert (await device.send_direct_message(stranger, "hello")).acked
     await device.disconnect()
 
 
@@ -1603,6 +1603,7 @@ class _ScriptedDevice:
     stand-in (any non-``None`` object counts as delivered) or ``None`` for an unacknowledged
     transmission — and appends the sent text to :attr:`sent`, so a test can assert exactly how
     many soft retries fired. Once the script is exhausted every further send goes unacked.
+    Each send gets its own ack code, as the radio's do.
     """
 
     def __init__(self, acks: list) -> None:
@@ -1613,8 +1614,11 @@ class _ScriptedDevice:
         return None
 
     async def send_direct_message(self, contact: Contact, text: str):
+        from meshterm.core.models import Delivery
+
         self.sent.append(text)
-        return self._acks.pop(0) if self._acks else None
+        ack = self._acks.pop(0) if self._acks else None
+        return Delivery(f"{len(self.sent):08x}", ack)
 
 
 def _contact() -> Contact:
@@ -2568,3 +2572,160 @@ def test_a_link_written_without_a_scheme_gets_its_code_too() -> None:
         assert isinstance(opened[0], QrScreen) and opened[0].urls == ("https://meshterm.net/map",)
 
     asyncio.run(scenario())
+
+
+# -- acks: listening early, waiting long enough, and catching the late ones ----
+
+
+class _AckingMeshCore:
+    """A meshcore stand-in whose send accepts the message, then acks it when scripted.
+
+    ``ack_at`` is seconds after the send returns (``None``: never); ``during_send`` pushes
+    the ack before the send has even returned, as a neighbour's can.
+    """
+
+    def __init__(self, *, ack_at: float | None, during_send: bool = False, flood: bool = True):
+        self.subs: list = []
+        self.commands = self
+        self._ack_at = ack_at
+        self._during = during_send
+        self._flood = flood
+
+    def subscribe(self, event_type, callback, attribute_filters=None):  # noqa: ANN001, ANN201
+        from types import SimpleNamespace
+
+        sub = SimpleNamespace(event_type=event_type, callback=callback)
+        sub.unsubscribe = lambda: self.subs.remove(sub) if sub in self.subs else None
+        self.subs.append(sub)
+        return sub
+
+    def _push_ack(self) -> None:
+        from types import SimpleNamespace
+
+        from meshcore import EventType
+
+        event = SimpleNamespace(type=EventType.ACK, payload={"code": "c0ffee01", "trip_time": 900})
+        for sub in list(self.subs):
+            if sub.event_type is EventType.ACK:
+                sub.callback(event)
+
+    async def send_msg(self, pub, text):  # noqa: ANN001, ANN201
+        from types import SimpleNamespace
+
+        if self._during:
+            self._push_ack()
+        elif self._ack_at is not None:
+            asyncio.get_running_loop().call_later(self._ack_at, self._push_ack)
+        return SimpleNamespace(
+            type="MSG_SENT",
+            is_error=lambda: False,
+            payload={
+                "type": 1 if self._flood else 0,
+                "expected_ack": bytes.fromhex("c0ffee01"),
+                "suggested_timeout": 100,  # 0.1 s: the radio's (too hopeful) estimate
+            },
+        )
+
+
+def _acking_device(mc) -> object:  # noqa: ANN001
+    from meshterm.core.connection import MeshCoreDevice
+
+    device = MeshCoreDevice(port="mock")
+    device._mc = mc
+    return device
+
+
+async def test_an_ack_quicker_than_the_send_is_still_caught() -> None:
+    """The listener is armed before the send, so a neighbour's instant ack isn't missed."""
+    device = _acking_device(_AckingMeshCore(ack_at=None, during_send=True))
+    delivery = await device.send_direct_message(_contact(), "hi")
+    assert delivery.acked and delivery.code == "c0ffee01" and delivery.flood is True
+
+
+async def test_an_ack_after_the_radios_estimate_is_still_waited_for(monkeypatch) -> None:  # noqa: ANN001
+    """THE bug: the wait was the radio's estimate × 1.2 — 0.12 s here, five on hardware.
+
+    A flood's ack rides home on the path the message found, and on hardware took twice the
+    estimate. The wait is now the estimate plus a round trip sized like a login's.
+    """
+    import meshterm.core.connection as connection
+
+    monkeypatch.setattr(connection, "trace_timeout", lambda hops: 0.5)
+    device = _acking_device(_AckingMeshCore(ack_at=0.3))
+    delivery = await device.send_direct_message(_contact(), "hi")
+    assert delivery.acked, "an ack at 0.3 s, after the old 0.12 s wait, counts"
+
+
+async def test_no_ack_in_the_wait_still_says_which_code_to_listen_for(monkeypatch) -> None:  # noqa: ANN001
+    """Not confirmed yet is not failed: the code goes back, for a late ack to settle."""
+    import meshterm.core.connection as connection
+
+    monkeypatch.setattr(connection, "trace_timeout", lambda hops: 0.05)
+    device = _acking_device(_AckingMeshCore(ack_at=None, flood=False))
+    delivery = await device.send_direct_message(_contact(), "hi")
+    assert not delivery.acked and delivery.code == "c0ffee01" and delivery.flood is False
+
+
+async def _late_ack_chat(tmp_path: Path, repo: Repository):  # noqa: ANN202
+    """A started chat service over a simulator whose acks for Alice arrive late."""
+    device = MockDevice()
+    device._late_acks.add("Alice")
+    ctx = _StubContext(device, repo)
+    chat = ChatService(ctx)
+    await chat.start()
+    alice = next(c for c in await device.get_contacts() if c.name == "Alice")
+    return device, chat, alice
+
+
+async def test_a_late_ack_turns_the_message_delivered(tmp_path: Path, repo: Repository) -> None:
+    """In history and in the very message the chat screen holds — its ✗ becomes ✓."""
+    device, chat, alice = await _late_ack_chat(tmp_path, repo)
+    heard: list[ChatMessage] = []
+    try:
+        sent = await chat.send_direct(alice, "you there?", on_late_ack=heard.append)
+        assert sent.acked is False
+        for _ in range(100):
+            if sent.acked:
+                break
+            await asyncio.sleep(0.01)
+        assert sent.acked is True
+        stored = repo.recent_chat_messages(is_channel=False, peer=alice.key_prefix)
+        assert stored[-1].acked is True
+        assert heard == [sent]
+    finally:
+        await chat.stop()
+        await device.disconnect()
+
+
+async def test_a_late_ack_for_an_earlier_attempt_settles_a_retried_message(
+    tmp_path: Path, repo: Repository
+) -> None:
+    """^R sends it again under a new code; the first attempt's ack still counts."""
+    device, chat, alice = await _late_ack_chat(tmp_path, repo)
+    try:
+        await chat.stop()  # hold the late acks back until both attempts are out
+        sent = await chat.send_direct(alice, "anyone?")
+        first = next(iter(chat._awaiting))
+        await chat.resend_direct(alice, sent)
+        assert len(chat._awaiting) == 2 and sent.acked is False
+        chat._settle_ack(first)
+        assert sent.acked is True and not chat._awaiting
+        stored = repo.recent_chat_messages(is_channel=False, peer=alice.key_prefix)
+        assert len(stored) == 1 and stored[0].acked is True
+    finally:
+        await device.disconnect()
+
+
+def test_only_the_acks_the_radio_could_still_push_are_awaited(repo: Repository) -> None:
+    """The radio keeps its last eight codes; a few more than that are kept here, no more."""
+    from meshterm.services.chat_service import _AWAITING_CAP
+
+    chat = ChatService(_StubContext(MockDevice(), repo))
+    messages = [ChatMessage(text=str(n), outbound=True, acked=False) for n in range(40)]
+    for n, message in enumerate(messages):
+        chat._await_ack(message, [f"{n:08x}"], None)
+    assert len(chat._awaiting) == _AWAITING_CAP
+    chat._settle_ack("00000000")  # long gone: nothing to settle
+    assert messages[0].acked is False
+    chat._settle_ack(f"{39:08x}")
+    assert messages[39].acked is True

tests/test_courier.py

@@ -40,11 +40,15 @@ class _StubChat:
     def __init__(self, outcomes: list[bool]) -> None:
         self.outcomes = list(outcomes)
         self.sent: list[tuple[str, str]] = []
+        self.late: list = []  # the late-ack listener each unacked send handed over
 
-    async def send_direct(self, contact: Contact, text: str) -> ChatMessage:
+    async def send_direct(self, contact: Contact, text: str, *, on_late_ack=None) -> ChatMessage:  # noqa: ANN001
         self.sent.append((contact.name, text))
         acked = self.outcomes.pop(0) if self.outcomes else False
-        return ChatMessage(text=text, outbound=True, acked=acked)
+        chat = ChatMessage(text=text, outbound=True, acked=acked)
+        if not acked:
+            self.late.append((chat, on_late_ack))
+        return chat
 
 
 class _StubDevice:
@@ -187,6 +191,26 @@ async def test_attempt_gives_up_after_the_budget(tmp_path: Path) -> None:
     assert len(gave) == 1
 
 
+async def test_a_late_ack_is_a_delivery_not_a_reason_to_send_again(tmp_path: Path) -> None:
+    """The ack outran the attempt's wait; when it lands, the entry is delivered.
+
+    It used to stay queued, and the next attempt sent the recipient the same message again —
+    on a mesh where most acks from beyond a neighbour arrive late, a duplicate per message.
+    """
+    service = _service(tmp_path, [False])
+    ctx = service._ctx
+    entry = ctx.courier_store.queue(NODE, "Hub", "hello")
+    await service._attempt(ctx.courier_store.get(entry.ident))
+    assert ctx.courier_store.get(entry.ident).status == QUEUED
+    chat, on_late_ack = ctx.chat.late[0]
+    on_late_ack(chat)
+    settled = ctx.courier_store.get(entry.ident)
+    assert settled.status == "delivered" and settled.attempts == 1
+    assert any("ack came late" in a.message for a in ctx.watch_store.alerts())
+    on_late_ack(chat)  # the radio can push the same ack twice
+    assert len([a for a in ctx.watch_store.alerts() if "delivered" in a.message]) == 1
+
+
 async def test_pass_attempts_at_most_one_entry(tmp_path: Path) -> None:
     """A backlog drains one message per pass — the courier never bursts."""
     service = _service(tmp_path, [True, True])

tests/test_rooms.py

@@ -468,7 +468,7 @@ async def test_the_simulators_room_sends_its_board_after_a_login(tmp_path: Path)
     )
     try:
         heard.clear()  # the stream's opening burst carries a synthetic DM of its own
-        assert await device.send_direct_message(ROOM, "before joining") is None
+        assert not (await device.send_direct_message(ROOM, "before joining")).acked
         login = await device.room_login(ROOM, "hello")
         assert login.access is RoomAccess.MEMBER
         while len([m for m in heard if m.is_post]) < 5:
@@ -478,7 +478,7 @@ async def test_the_simulators_room_sends_its_board_after_a_login(tmp_path: Path)
         assert [m.sender_timestamp for m in posts] == sorted(m.sender_timestamp for m in posts)
         assert {m.author for m in posts} == {ALICE_KEY, "c3d4e5f6", STRANGER_KEY, "f6a7b8c9"}
 
-        assert await device.send_direct_message(ROOM, "mine") is not None  # kept, acked
+        assert (await device.send_direct_message(ROOM, "mine")).acked  # kept, acked
         heard.clear()
         await device.room_login(ROOM, "")  # the room knows us; nothing new but our own
         await asyncio.sleep(0.2)
@@ -494,7 +494,7 @@ async def test_a_read_only_member_is_heard_but_not_kept(tmp_path: Path) -> None:
     device._remote_config(ROOM)["allow.read.only"] = "on"
     login = await device.room_login(ROOM, "anything")
     assert login.access is RoomAccess.READ_ONLY
-    assert await device.send_direct_message(ROOM, "lost") is None
+    assert not (await device.send_direct_message(ROOM, "lost")).acked
     await device.disconnect()