A direct message's frames are its own, by the MAC over its ciphertext

The paths on a DM listed a message's own retransmissions, the other end's
traffic, and both of those for every other message sent in the same minute and
a half. On one real bob_caribou message that was 30 rows over 8 paths, of
which 23 were the same two one-hop routes repeated. Those one-hops were the
tell (JP): they were our *own* sends, overheard coming back off the two
repeaters in earshot, correctly showing one hop because one hop is all they had
travelled — but shown under the message as if they were its route.

The channel view has always matched by decrypted content, and asked for here.
A direct frame cannot be read, but it does not need to be: its body is
[dest 1B][src 1B][MAC 2B][ciphertext], and that MAC is a tag over this
message's plaintext. Copies of one message share it; the next message's differ.
frame_addressing already knew where those two bytes were and threw them away,
and the observations table already had a cipher_mac column its channel sibling
used — so a direct message now groups exactly as a channel one does, with no
schema change and nothing decrypted.

Three filters, in order. The frame ran between the conversation's two ends; it
travelled the direction this message did; it carries its group's MAC. The
middle one is doing more work than it looks: both ends' hashes sit on every
frame either way, so matching on the pair alone was what put our sends under a
received message.

The window is clamped by the conversation's own messages rather than a flat
±90 s — nine messages fell inside one such window. The clamp reads only
messages going the *same* way, which the first cut of it got wrong: an inbound
stamp is the sender's clock, so ordering it against our own sends compares two
clocks, and a received message came back with an empty window. The directions
are also bounded differently, because we know different things about them. A
send is stamped by our clock as it leaves, so no frame of it predates it and
all the room goes forward to the next send, where the retries are. A received
message keeps a centred window, halfway to the received messages either side.

Rows are paths now, not frames: identical routes fold into one line carrying
its copy count, its first sighting, and the best SNR that path managed. Twelve
identical lines said "twelve arrivals" where the truth was "two paths, six
times each".

Measured on the conversation that prompted this, whose history predates the
MAC and so exercises only the direction and window fixes:

  outbound "Allo! Ça marche JF"   30 rows / 8 paths -> 2 rows (x3 each)
  inbound  "J'ai pas de ack..."   30 rows / 8 paths -> 2 rows, 6 and 7 hops

which is exactly what the packet log says by hand. Old history keeps the
address-and-time fallback and is billed as such; frames recorded from now on
match exactly, and say so in the same words the channel view uses.

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

meshterm/core/frames.py

@@ -97,6 +97,9 @@ def frame_addressing(payload: Mapping[str, Any]) -> dict[str, str]:
 
     * ``dest_hash`` — the recipient's one-byte key hash (every addressed class).
     * ``src_hash`` — the sender's one-byte key hash (the two-hash classes).
+    * ``cipher_mac`` — the two-byte MAC that follows them: a tag over the encrypted
+      message, and so a fingerprint identifying *which* message a frame carries without
+      being able to read it.
     * ``src_key`` — the sender's *whole* public key, which an anonymous request carries
       instead of a hash (it has no shared secret to be recognised by yet).
     * ``ack_crc`` — an ack's four-byte checksum of the message it acknowledges, hex in
@@ -137,6 +140,13 @@ def frame_addressing(payload: Mapping[str, Any]) -> dict[str, str]:
         return {
             "dest_hash": body[:hash_w].hex(),
             "src_hash": body[hash_w : hash_w * 2].hex(),
+            # The MAC an addressed frame carries is a tag over *this* message's plaintext,
+            # so two frames sharing one (between the same pair) are copies of the same
+            # message — a fingerprint that needs no key to compare. It is what lets the
+            # message-paths view group a direct message's own retransmissions and tell them
+            # from the next message's, which content-matching does for a channel frame we
+            # can decrypt (see :mod:`~meshterm.services.message_paths`).
+            "cipher_mac": body[hash_w * 2 : hash_w * 2 + _MAC].hex(),
         }
     if typename == "ANON_REQ":
         if len(body) < hash_w + _KEY + _MAC:

meshterm/persistence/repository.py

@@ -14,7 +14,7 @@ from datetime import datetime, timedelta
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Optional, Sequence
 
-from ..core.frames import CHANNEL_CLASSES, ENDPOINT_HASH_BYTES
+from ..core.frames import ADDRESSED_CLASSES, CHANNEL_CLASSES, ENDPOINT_HASH_BYTES
 from ..core.models import (
     PATH_TRACE_TARGET,
     ChatMessage,
@@ -88,11 +88,12 @@ def _packet_raw(row: sqlite3.Row) -> Optional[dict]:
     """
     typename = _row_value(row, "payload_typename")
     chan_hash = row["chan_hash"]
+    cipher_mac = _row_value(row, "cipher_mac")
     dest = _row_value(row, "dest")
     src = _row_value(row, "src")
     tag = _row_value(row, "tag")
     trace_snrs = _row_value(row, "trace_snrs")
-    if not any((typename, chan_hash, dest, src, tag, trace_snrs)):
+    if not any((typename, chan_hash, cipher_mac, dest, src, tag, trace_snrs)):
         return None
     raw: dict = {}
     if typename:
@@ -102,6 +103,10 @@ def _packet_raw(row: sqlite3.Row) -> Optional[dict]:
         raw["chan_hash"] = chan_hash
         raw["cipher_mac"] = row["cipher_mac"]
         raw["crypted"] = row["crypted"]
+    if cipher_mac and not chan_hash:
+        # An addressed frame keeps only its MAC (no channel envelope), so it is restored
+        # here rather than in the chan_hash branch above.
+        raw["cipher_mac"] = cipher_mac
     if dest:
         raw["dest_hash"] = dest
     if src:
@@ -114,6 +119,16 @@ def _packet_raw(row: sqlite3.Row) -> Optional[dict]:
     return raw
 
 
+def _as_when(iso: Any) -> Optional[datetime]:
+    """Parse a stored ISO timestamp, or ``None`` when absent/unparseable."""
+    if not iso:
+        return None
+    try:
+        return datetime.fromisoformat(str(iso))
+    except (TypeError, ValueError):
+        return None
+
+
 def _row_value(row: sqlite3.Row, key: str) -> Any:
     """Read ``key`` from a row, tolerating a query that didn't select it (returns ``None``)."""
     try:
@@ -992,6 +1007,13 @@ class Repository:
             chan_hash = raw.get("chan_hash")
             cipher_mac = raw.get("cipher_mac")
             crypted = raw.get("crypted")
+        elif typename in ADDRESSED_CLASSES:
+            # An addressed frame's MAC, in the same column its channel sibling uses: a tag
+            # over the encrypted body, so copies of one message share it and the next
+            # message's do not. The ciphertext itself is *not* kept — we could never read
+            # it, and only the fingerprint is needed to tell one message's frames apart
+            # from another's (see :mod:`~meshterm.services.message_paths`).
+            cipher_mac = raw.get("cipher_mac")
         # One column each for the three shapes of addressing: the sender is a hash or a
         # whole key (its length says which), the token an ack's checksum or a trace's tag
         # (the payload class says which) — so neither needs a column of its own.
@@ -1929,6 +1951,56 @@ class Repository:
         ).fetchall()
         return {row["peer"]: datetime.fromisoformat(row["last"]) for row in rows}
 
+    def direct_message_bounds(
+        self, peer: Optional[str], when: datetime, *, outbound: bool
+    ) -> "tuple[Optional[datetime], Optional[datetime]]":
+        """The times of the direct messages either side of ``when`` in one conversation.
+
+        What bounds the message-paths search for a direct message (see
+        :mod:`~meshterm.services.message_paths`). A direct frame is encrypted, so the log
+        cannot say which message it carried; a fixed window around the message is therefore
+        the only bound available — and at conversational pace it is far too wide. Nine
+        messages of one real exchange fell inside a single ±90 s window, so every one of
+        them listed all nine messages' frames as its own.
+
+        The messages around it are the natural edges: a frame logged after the *next*
+        message was composed belongs to that one, not this one. The caller clamps its window
+        to the midpoint of each gap, so the bound tightens exactly as the conversation
+        speeds up.
+
+        Only messages travelling the **same way** count as neighbours, because the two
+        directions are stamped by two different clocks: our own sends carry ours, taken as
+        they leave, while a received message carries the *sender's*, which may drift from
+        ours by minutes. Ordering an inbound message against an outbound one therefore
+        compares two clocks and can bound a message by an edge that, in its own clock,
+        hasn't happened yet — which is exactly how the first cut of this clamp gave a
+        received message an empty window. Within one direction the stamps are consistent,
+        so the neighbours mean what they say.
+
+        Args:
+            peer: The contact's key prefix, as :meth:`record_chat_message` stores it.
+            when: The message's own time.
+            outbound: The direction to look along — the message's own.
+
+        Returns:
+            ``(previous, next)`` same-direction message times, either of which is ``None``
+            when this message is the first or last of its side of the conversation.
+        """
+        key = (peer or "").lower()
+        args = (key, int(outbound), when.isoformat())
+        row = self._conn.execute(
+            "SELECT MAX(created_at) AS edge FROM messages "
+            "WHERE is_channel = 0 AND peer = ? AND outbound = ? AND created_at < ?",
+            args,
+        ).fetchone()
+        prev = _as_when(row["edge"] if row else None)
+        row = self._conn.execute(
+            "SELECT MIN(created_at) AS edge FROM messages "
+            "WHERE is_channel = 0 AND peer = ? AND outbound = ? AND created_at > ?",
+            args,
+        ).fetchone()
+        return prev, _as_when(row["edge"] if row else None)
+
     def channel_stats(self) -> dict[str, ChannelStats]:
         """Aggregate stored channel messages into per-channel statistics.
 

meshterm/services/message_paths.py

@@ -14,21 +14,28 @@ chat message so the chat screen can show, on demand, every way the message reach
   ``Name: `` sender prefix convention on either side) is one arrival of that message —
   our own broadcasts included, since a repeater's rebroadcast of us is overheard and
   logged like anything else.
-* **Direct messages** ride ECDH-encrypted ``TEXT_MSG`` frames that only the recipient
-  can decrypt, so no content match is possible from the log. What such a frame does
-  carry in the clear is its *addressing* — the one-byte key hash of each end
-  (:mod:`~meshterm.core.frames`) — so the window narrows to the frames travelling
-  between this conversation's two ends, in either direction, and what is left is billed
-  as matched by address and time. Honest evidence, not a claim: a one-byte hash
-  collides, and a frame between the right pair in the right ninety seconds is still
-  only *probably* this message.
+* **Direct messages** ride ECDH-encrypted ``TEXT_MSG`` frames that only the recipient can
+  decrypt, so no *content* match is possible from the log. What such a frame carries in
+  the clear is its **addressing and its MAC** (:mod:`~meshterm.core.frames`): the one-byte
+  key hash of each end, then a two-byte tag over the encrypted body. That MAC is a
+  fingerprint of the message itself — copies of one message share it and the next message's
+  do not — so direct frames group as exactly as channel frames do, without ever reading
+  them. Three filters, in order: the frame ran **between this conversation's two ends**,
+  in the **direction this message travelled** (our own sends are ``us → peer``; a received
+  message is ``peer → us``), and it carries **the same MAC** as the rest of its group.
+
+  A conversation's messages are then told apart by which MAC group sits nearest the
+  message's own time. That leaves one honest gap: history recorded before the MAC was kept
+  has none, and those messages fall back to matching by address, direction and time —
+  billed as such, because a frame between the right pair in the right window is only
+  *probably* this one.
 
 Nothing here transmits; it is a read-model over the repository.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime, timedelta
 from typing import TYPE_CHECKING, Optional, Sequence
 
@@ -43,10 +50,17 @@ if TYPE_CHECKING:
 #: message is stamped with the *sender's* clock, which may drift from ours by minutes.
 _CHANNEL_WINDOW = timedelta(minutes=15)
 
-#: How far around a direct message the log is searched. Tight, because the frame's
-#: addressing and the clock are the only evidence tying it to the message.
+#: How far around a direct message the log is searched, before the conversation's own
+#: messages narrow it further (see :func:`direct_window`). Generous on its own: a send is
+#: retried for a while after it is composed, and an inbound message is stamped with the
+#: sender's clock.
 _DIRECT_WINDOW = timedelta(seconds=90)
 
+#: The least room left either side of a direct message once its neighbours clamp the
+#: window. Two messages composed a second apart still each need somewhere to look, and the
+#: midpoint between them would otherwise leave one of them nothing at all.
+_MIN_HALF_WINDOW = timedelta(seconds=2)
+
 #: Frame classes that carry a direct (addressed) text message — the payload class the
 #: meshcore library names, spelled exactly as it reports it (see
 #: :data:`~meshterm.core.frames.ADDRESSED_CLASSES`, the packet viewer's lane, the
@@ -69,12 +83,19 @@ class Arrival:
         snr: Reception SNR in dB — of the *last relay*, as with every packet row.
         resend: The sender's resend counter for this copy (0 = the original send),
             recovered from a decrypted channel frame; always 0 for direct frames.
+        copies: How many logged frames this row stands for — always ``1`` as the matchers
+            produce them, and more once :func:`collapse` folds a path heard several times
+            into one row. A direct send is retried, and each retry is overheard off every
+            repeater in earshot, so one message routinely leaves a dozen frames on two
+            paths; listing them individually said "twelve arrivals" where the truth was
+            "two paths, six times each".
     """
 
     when: datetime
     hops: tuple[str, ...]
     snr: Optional[float]
     resend: int = 0
+    copies: int = 1
 
 
 def _frame_hops(observation: Observation) -> tuple[str, ...]:
@@ -154,51 +175,178 @@ def _endpoint_hash(key: Optional[str]) -> str:
     return text[:_HASH_CHARS] if len(text) >= _HASH_CHARS else ""
 
 
-def direct_frames_near(
+def direct_window(
+    repo: "Repository", message: ChatMessage
+) -> tuple[datetime, datetime]:
+    """The span of log to search for one direct message's frames.
+
+    :data:`_DIRECT_WINDOW` either side, then clamped by the conversation's own messages —
+    only those travelling the **same way**, because the two directions are stamped by two
+    different clocks (see :meth:`~meshterm.persistence.repository.Repository
+    .direct_message_bounds`). Without any clamp the flat window is wildly too wide at
+    conversational pace: nine messages of one real exchange fell inside a single ±90 s
+    window, so each of them claimed all nine messages' frames as its own.
+
+    The two directions are clamped differently, because we know very different things about
+    them:
+
+    * **Our own sends** are stamped as they leave, by our own clock, so no frame of one can
+      predate it — the window opens *at* the message (less a moment for jitter) and runs to
+      the next send. Retries follow the send, so all the room goes forward.
+    * **A received message** carries the sender's clock, which may sit either side of when
+      we actually heard it. Its window stays centred, reaching halfway to the messages
+      before and after it.
+
+    Never narrower than :data:`_MIN_HALF_WINDOW` either side, so two messages a second apart
+    still each have somewhere to look.
+
+    Args:
+        repo: The repository holding the messages.
+        message: The message to bound a search around.
+
+    Returns:
+        ``(start, end)`` for :meth:`~meshterm.persistence.repository.Repository
+        .packet_frames_between`.
+    """
+    at = message.created_at
+    start, end = at - _DIRECT_WINDOW, at + _DIRECT_WINDOW
+    previous, following = repo.direct_message_bounds(
+        message.peer, at, outbound=message.outbound
+    )
+    if message.outbound:
+        # Our clock, so nothing of this send predates it; the whole window goes forward,
+        # up to the next send.
+        start = at - _MIN_HALF_WINDOW
+        if following is not None:
+            end = min(end, max(at + _MIN_HALF_WINDOW, following))
+    else:
+        if previous is not None:
+            start = max(start, min(at - _MIN_HALF_WINDOW, previous + (at - previous) / 2))
+        if following is not None:
+            end = min(end, max(at + _MIN_HALF_WINDOW, at + (following - at) / 2))
+    return start, end
+
+
+def _addressed(raw: dict) -> tuple[str, str]:
+    """A frame's ``(source, destination)`` endpoint hashes, lowercased (empty when absent)."""
+    return (
+        str(raw.get("src_hash") or "").lower(),
+        str(raw.get("dest_hash") or "").lower(),
+    )
+
+
+def _pick_group(
+    groups: "dict[str, list[Arrival]]", message: ChatMessage
+) -> list[Arrival]:
+    """The MAC group that is this message, out of the ones in the window.
+
+    Each group is one message's frames — same MAC, so the same encrypted body. Which of
+    them is *this* message is the one thing the log cannot say, so the clock decides, and it
+    decides well: a send's frames start at the moment it was composed and run on through the
+    retries, so the group to take is the earliest one that did not start before the message
+    did. Falling back to the nearest group covers an inbound message, whose stamp is the
+    sender's clock and may sit slightly after the frame we heard.
+    """
+    if not groups:
+        return []
+    at = message.created_at
+    started = {mac: min(a.when for a in arrivals) for mac, arrivals in groups.items()}
+    after = [mac for mac, first in started.items() if first >= at]
+    if after:
+        return groups[min(after, key=lambda mac: started[mac])]
+    return groups[min(started, key=lambda mac: abs(started[mac] - at))]
+
+
+def direct_arrivals(
     repo: "Repository",
     message: ChatMessage,
     *,
-    ends: Sequence[Optional[str]] = (),
-) -> list[Arrival]:
-    """Direct-message frames logged around ``message``, matched by address and time.
+    self_key: Optional[str] = None,
+    peer_key: Optional[str] = None,
+) -> tuple[list[Arrival], bool]:
+    """Every logged frame of one direct message, and whether they were matched exactly.
 
-    Direct frames are encrypted to their recipient, so the log can't confirm which
-    message a frame carried — the caller must present these as correlated evidence, not
-    a claim (see the module docstring). Their *addressing* is in the clear, though, so
-    naming the conversation's two ends narrows the window to the frames that ran between
-    those two nodes, in either direction, instead of every direct frame the radio
-    happened to overhear. A frame whose addressing wasn't recovered can't be shown to
-    belong, so it drops out — evidence, not guesswork.
+    Three filters (see the module docstring): the frame ran between this conversation's two
+    ends, in the direction this message travelled, and — where the log kept it — it shares
+    a MAC with the rest of its group, which is what makes one message's frames separable
+    from the next message's without decrypting either.
+
+    **Direction matters more than it looks.** Both ends' hashes appear on every frame of a
+    conversation, whichever way it went, so matching on the pair alone put our own outgoing
+    frames into the view of a message we *received* and vice versa. That is where the
+    implausible one-hop rows came from: they were our own sends, overheard coming back off
+    the two repeaters in earshot, correctly showing one hop because one hop is all they had
+    travelled — but shown under an inbound message as if they were its route to us.
 
     Args:
         repo: The repository holding the packet log.
-        message: The chat message to search around.
-        ends: Keys (or key prefixes) of the conversation's two ends — the peer and our
-            own node. Each contributes its endpoint hash; whichever are given must
-            *both* appear on a frame for it to count. Empty falls back to every direct
-            frame in the window, matched by time alone.
+        message: The chat message whose frames to find.
+        self_key: Our own node's key (or prefix) — one end of every frame here.
+        peer_key: The conversation peer's key (or prefix) — the other end.
 
     Returns:
-        The matching frames as arrivals, oldest first.
+        ``(arrivals, exact)`` — the frames oldest first, and whether a MAC identified them
+        as one message rather than merely correlating them by address and time.
     """
-    wanted = {h for h in (_endpoint_hash(end) for end in ends) if h}
-    frames = repo.packet_frames_between(
-        message.created_at - _DIRECT_WINDOW, message.created_at + _DIRECT_WINDOW
-    )
-    arrivals: list[Arrival] = []
-    for frame in frames:
+    ours, theirs = _endpoint_hash(self_key), _endpoint_hash(peer_key)
+    # Which way this message travelled, so a send's own frames and a received message's
+    # never appear under each other.
+    if message.outbound:
+        want_src, want_dest = ours, theirs
+    else:
+        want_src, want_dest = theirs, ours
+
+    start, end = direct_window(repo, message)
+    groups: dict[str, list[Arrival]] = {}
+    loose: list[Arrival] = []
+    for frame in repo.packet_frames_between(start, end):
         raw = frame.raw if isinstance(frame.raw, dict) else {}
         if raw.get("payload_typename") not in _DIRECT_TYPENAMES:
             continue
-        if wanted:
-            addressed = {
-                str(raw.get("dest_hash") or "").lower(),
-                str(raw.get("src_hash") or "").lower(),
-            }
-            if not wanted <= addressed:
-                continue
-        arrivals.append(Arrival(when=frame.observed_at, hops=_frame_hops(frame), snr=frame.snr))
-    return arrivals
+        src, dest = _addressed(raw)
+        if (want_src and src != want_src) or (want_dest and dest != want_dest):
+            continue
+        arrival = Arrival(when=frame.observed_at, hops=_frame_hops(frame), snr=frame.snr)
+        mac = str(raw.get("cipher_mac") or "").lower()
+        if mac:
+            groups.setdefault(mac, []).append(arrival)
+        else:
+            loose.append(arrival)
+    if groups:
+        return _pick_group(groups, message), True
+    # History recorded before the MAC was kept: address, direction and time are all there
+    # is, and the caller says so rather than claiming more.
+    return loose, False
+
+
+def collapse(arrivals: Sequence[Arrival]) -> list[Arrival]:
+    """Fold arrivals that rode the same path into one row apiece, counting the copies.
+
+    A direct send is retried and each retry is overheard off every repeater in earshot, so
+    one message leaves a dozen frames on two paths. The graph has always drawn the *distinct*
+    paths; the row list showed every frame, which read as twelve arrivals where the truth was
+    two paths heard six times each. Each surviving row keeps its **first** sighting's time
+    and its **best** SNR — the earliest is when the path first worked, and the strongest is
+    what the path is capable of.
+
+    Args:
+        arrivals: The matched frames, oldest first.
+
+    Returns:
+        One :class:`Arrival` per distinct path, in first-heard order, each carrying its
+        :attr:`~Arrival.copies` count.
+    """
+    folded: dict[tuple[str, ...], Arrival] = {}
+    for arrival in arrivals:
+        seen = folded.get(arrival.hops)
+        if seen is None:
+            folded[arrival.hops] = replace(arrival, copies=1)
+            continue
+        best = seen.snr if arrival.snr is None else (
+            arrival.snr if seen.snr is None else max(seen.snr, arrival.snr)
+        )
+        folded[arrival.hops] = replace(seen, snr=best, copies=seen.copies + 1)
+    return list(folded.values())
 
 
 def distinct_paths(arrivals: list[Arrival]) -> int:

meshterm/ui/chat.py

@@ -1182,7 +1182,8 @@ async def _make_paths_presenter(
     from ..services import trace_runner
     from ..services.message_paths import (
         channel_arrivals,
-        direct_frames_near,
+        collapse,
+        direct_arrivals,
         distinct_paths,
     )
     from .message_paths_screen import MessagePathsScreen
@@ -1201,7 +1202,7 @@ async def _make_paths_presenter(
         info = await ctx.devstate.self_info()
         self_name = str(info.get("name") or "") or None
         # Our own key names one end of every direct frame in this conversation; the
-        # contact's key names the other (see :func:`direct_frames_near`).
+        # contact's key names the other (see :func:`direct_arrivals`).
         self_key = str(info.get("public_key") or "") or None
     except Exception:  # noqa: BLE001 - a nameless self just skips the white highlight
         self_name = None
@@ -1242,13 +1243,27 @@ async def _make_paths_presenter(
             peer_key = conversation.peer or (
                 (contact.public_key or contact.key_prefix) if contact else None
             )
-            arrivals = direct_frames_near(ctx.repo, message, ends=(peer_key, self_key))
-            matched = False
-            summary = (
-                "direct frames are encrypted — matched by address and time (±90 s)"
-                if peer_key and self_key
-                else "direct frames are encrypted — matched by time alone (±90 s)"
+            arrivals, exact = direct_arrivals(
+                ctx.repo, message, self_key=self_key, peer_key=peer_key
             )
+            # An exact match is the same claim the channel view makes: these frames are
+            # this message. It is earned by the frames' shared MAC, so it survives being
+            # unable to read a 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 graph's left endpoint: who the message set out from. Our own sends are
         # us; an inbound channel message names its sender on the wire; a direct chat's
         # origin is the conversation's peer (billed as time-matched by the summary).

meshterm/ui/message_paths_screen.py

@@ -286,10 +286,19 @@ class MessagePathsScreen(Screen):
         The hop count leads (:func:`~meshterm.ui.pathline.hops_atom`): it is the one fact
         about the route the line above encodes without stating, and the number two
         arrivals of the same message are compared on before anything else.
+
+        A row stands for a *path*, not a frame (see
+        :func:`~meshterm.services.message_paths.collapse`), so a path heard more than once
+        says how many times and the time reads as the first of them — a direct send is
+        retried and every retry is overheard off every repeater in earshot, which used to
+        fill the list with a dozen identical lines. The SNR is the best that path managed,
+        which is what it is capable of.
         """
         text = hops_atom(len(arrival.hops))
         text.append("  ")
         text.append(arrival.when.astimezone().strftime("%H:%M:%S"), style="muted")
+        if arrival.copies > 1:
+            text.append(f"  ×{arrival.copies}", style="muted")
         if arrival.snr is not None:
             text.append("  ")
             text.append(f"{arrival.snr:+.1f} dB", style=snr_style(arrival.snr))

tests/test_frames.py

@@ -29,15 +29,22 @@ def _frame(typename: str, body: bytes, **extra) -> dict:
     return {"payload_typename": typename, "pkt_payload": body, **extra}
 
 
-def test_addressed_classes_name_both_ends() -> None:
-    """A direct message, request, response or returned path opens with dest then src."""
+def test_addressed_classes_name_both_ends_and_their_mac() -> None:
+    """A direct message, request, response or returned path opens dest, src, then MAC.
+
+    The MAC is a tag over the encrypted body, so it fingerprints *which* message a frame
+    carries without the frame ever being read — which is how the message-paths view groups
+    a direct message's own retransmissions and tells them from the next message's (see
+    :mod:`~meshterm.services.message_paths`).
+    """
     body = bytes.fromhex("3da1") + b"\xab\xcd" + b"\x00" * 16
     for typename in ("TEXT_MSG", "REQ", "RESPONSE", "PATH"):
         assert frame_addressing(_frame(typename, body)) == {
-            "dest_hash": "3d", "src_hash": "a1",
+            "dest_hash": "3d", "src_hash": "a1", "cipher_mac": "abcd",
         }
 
 
+
 def test_anonymous_request_carries_its_senders_whole_key() -> None:
     """No shared secret yet, so the sender identifies itself in full — 32 bytes of key."""
     key = bytes(range(32))

tests/test_message_paths.py

@@ -19,7 +19,8 @@ from meshterm.core.models import ChatMessage, Observation, utcnow
 from meshterm.persistence.repository import Repository
 from meshterm.services.message_paths import (
     channel_arrivals,
-    direct_frames_near,
+    collapse,
+    direct_arrivals,
     distinct_paths,
 )
 
@@ -113,63 +114,169 @@ def test_channel_arrivals_ignore_frames_outside_the_window(tmp_path: Path) -> No
     repo.close()
 
 
-def _direct_raw(dest: str = "", src: str = "") -> dict:
+def _direct_raw(dest: str = "", src: str = "", mac: str = "") -> dict:
     """A direct-message frame's raw payload, spelled as the meshcore library reports it."""
     raw: dict = {"payload_typename": "TEXT_MSG"}
     if dest:
         raw["dest_hash"] = dest
     if src:
         raw["src_hash"] = src
+    if mac:
+        raw["cipher_mac"] = mac
     return raw
 
 
-def test_direct_frames_match_by_time_only(tmp_path: Path) -> None:
-    """With neither end named, every direct frame in the tight window is the evidence."""
+def test_direct_frames_without_a_mac_fall_back_to_address_and_time(tmp_path: Path) -> None:
+    """History recorded before the MAC was kept still correlates — and says that it did.
+
+    The fallback is honest evidence, not a claim: a frame between the right pair in the
+    right window is only *probably* this message, and ``exact`` comes back ``False`` so the
+    view can bill it that way.
+    """
     repo, run = _repo(tmp_path)
     now = utcnow()
-    _record_frame(repo, run, when=now + timedelta(seconds=3), path="3d63", raw=_direct_raw())
+    _record_frame(
+        repo, run, when=now + timedelta(seconds=3), path="3d63",
+        raw=_direct_raw(dest="d4", src="a1"),
+    )
     _record_frame(  # a channel frame in the window is not direct-message evidence
         repo, run, when=now + timedelta(seconds=4), path="",
         raw=_grp_txt_raw(SECRET, "Alice: hi"),
     )
-    _record_frame(  # a direct frame far outside the tight window doesn't correlate
-        repo, run, when=now + timedelta(minutes=10), path="", raw=_direct_raw(),
-    )
-    message = ChatMessage(text="see you at 8", peer="d4e5", created_at=now)
-    arrivals = direct_frames_near(repo, message)
-    assert len(arrivals) == 1 and arrivals[0].hops == ("3d63",)
-    repo.close()
-
-
-def test_direct_frames_narrow_to_the_conversation_ends(tmp_path: Path) -> None:
-    """Naming both ends keeps only the frames that ran between them, either direction."""
-    repo, run = _repo(tmp_path)
-    now = utcnow()
-    # Us (a1…) and the peer (d4…), each way round: both are this conversation's traffic.
-    _record_frame(
-        repo, run, when=now + timedelta(seconds=1), path="3d63",
+    _record_frame(  # a direct frame far outside the window doesn't correlate
+        repo, run, when=now + timedelta(minutes=10), path="",
         raw=_direct_raw(dest="d4", src="a1"),
     )
-    _record_frame(
-        repo, run, when=now + timedelta(seconds=2), path="c0",
-        raw=_direct_raw(dest="a1", src="d4"),
+    message = ChatMessage(
+        text="see you at 8", outbound=True, peer="d4e5", created_at=now
     )
-    # Someone else's direct message, overheard in the same window.
-    _record_frame(
-        repo, run, when=now + timedelta(seconds=3), path="3d63",
-        raw=_direct_raw(dest="7f", src="c0"),
+    arrivals, exact = direct_arrivals(
+        repo, message, self_key="a1" + "0" * 62, peer_key="d4e5"
     )
-    # Addressed to us, but by a third party — not this conversation.
-    _record_frame(
-        repo, run, when=now + timedelta(seconds=4), path="",
-        raw=_direct_raw(dest="a1", src="7f"),
-    )
-    # Its addressing was never recovered, so it can't be shown to belong.
-    _record_frame(repo, run, when=now + timedelta(seconds=5), path="", raw=_direct_raw())
-
-    message = ChatMessage(text="see you at 8", peer="d4e5f6a7", created_at=now)
-    arrivals = direct_frames_near(
-        repo, message, ends=("d4e5f6a7", "a1" + "0" * 62),
-    )
-    assert [a.hops for a in arrivals] == [("3d63",), ("c0",)]
+    assert [a.hops for a in arrivals] == [("3d63",)]
+    assert exact is False
     repo.close()
+
+
+def test_direct_frames_keep_only_the_direction_the_message_travelled(tmp_path: Path) -> None:
+    """A send shows our outgoing frames; a received message shows the incoming ones.
+
+    Both ends' hashes sit on every frame of a conversation whichever way it went, so
+    matching on the pair alone put our own sends into a received message's view. That is
+    where the implausible one-hop rows came from (JP, 2026-09-02): our transmissions heard
+    coming back off the repeaters in earshot, correctly one hop, shown as if they were an
+    inbound route to us.
+    """
+    repo, run = _repo(tmp_path)
+    now = utcnow()
+    ours = dict(dest="d4", src="a1", mac="beef")     # us -> peer
+    theirs = dict(dest="a1", src="d4", mac="f00d")   # peer -> us
+    _record_frame(repo, run, when=now + timedelta(seconds=1), path="3d63",
+                  raw=_direct_raw(**ours))
+    _record_frame(repo, run, when=now + timedelta(seconds=2), path="c0",
+                  raw=_direct_raw(**theirs))
+    # Someone else's traffic, overheard in the same window.
+    _record_frame(repo, run, when=now + timedelta(seconds=3), path="3d63",
+                  raw=_direct_raw(dest="7f", src="c0", mac="dead"))
+    _record_frame(repo, run, when=now + timedelta(seconds=4), path="",
+                  raw=_direct_raw(dest="a1", src="7f", mac="cafe"))
+
+    sent = ChatMessage(text="see you at 8", outbound=True, peer="d4e5f6a7", created_at=now)
+    got = ChatMessage(text="ok", outbound=False, peer="d4e5f6a7", created_at=now)
+    keys = dict(self_key="a1" + "0" * 62, peer_key="d4e5f6a7")
+
+    assert [a.hops for a in direct_arrivals(repo, sent, **keys)[0]] == [("3d63",)]
+    assert [a.hops for a in direct_arrivals(repo, got, **keys)[0]] == [("c0",)]
+    repo.close()
+
+
+def test_a_shared_mac_separates_one_message_from_the_next(tmp_path: Path) -> None:
+    """Two sends seconds apart keep their own frames — the MAC is the fingerprint.
+
+    This is the direct-message counterpart of the channel view's content match: the frames
+    cannot be read, but copies of one message carry the same MAC over its ciphertext and the
+    next message's carry a different one.
+    """
+    repo, run = _repo(tmp_path)
+    now = utcnow()
+    # First send, retried twice and heard off two repeaters.
+    for offset, path in ((1, "3d63"), (1, "27d4"), (6, "3d63"), (6, "27d4")):
+        _record_frame(repo, run, when=now + timedelta(seconds=offset), path=path,
+                      raw=_direct_raw(dest="d4", src="a1", mac="beef"))
+    # Second send, twelve seconds later, same pair and same paths.
+    for offset, path in ((13, "3d63"), (13, "27d4")):
+        _record_frame(repo, run, when=now + timedelta(seconds=offset), path=path,
+                      raw=_direct_raw(dest="d4", src="a1", mac="f00d"))
+
+    keys = dict(self_key="a1" + "0" * 62, peer_key="d4e5f6a7")
+    first = ChatMessage(text="one", outbound=True, peer="d4e5f6a7", created_at=now)
+    second = ChatMessage(
+        text="two", outbound=True, peer="d4e5f6a7",
+        created_at=now + timedelta(seconds=12),
+    )
+    got, exact = direct_arrivals(repo, first, **keys)
+    assert exact is True
+    assert len(got) == 4, "the first send's four frames, and not the second's"
+    assert len(direct_arrivals(repo, second, **keys)[0]) == 2
+    repo.close()
+
+
+def test_the_window_clamps_to_the_messages_going_the_same_way(tmp_path: Path) -> None:
+    """Each direction is bounded by its own neighbours, and bounded differently.
+
+    Without a clamp a fast exchange put every message's frames in every other's view — nine
+    messages of one real conversation fell inside a single flat window. With the wrong
+    clamp, a *received* message got an empty one: its stamp is the sender's clock, so
+    ordering it against our own sends compares two clocks and can bound it by an edge that,
+    in its own clock, hasn't happened yet (JP, 2026-09-02).
+    """
+    from meshterm.services.message_paths import _DIRECT_WINDOW, direct_window
+
+    repo, run = _repo(tmp_path)
+    now = utcnow()
+    sent = [now, now + timedelta(seconds=20)]
+    received = [now - timedelta(seconds=30), now + timedelta(seconds=50)]
+    for at in sent:
+        repo.record_chat_message(
+            ChatMessage(text="ours", outbound=True, peer="d4e5f6a7", created_at=at)
+        )
+    for at in received:
+        repo.record_chat_message(
+            ChatMessage(text="theirs", outbound=False, peer="d4e5f6a7", created_at=at)
+        )
+
+    # Our own send: our clock, so nothing of it predates it — the window opens at the
+    # message and runs forward to the next send, where retries actually live.
+    ours = ChatMessage(text="ours", outbound=True, peer="d4e5f6a7", created_at=now)
+    start, end = direct_window(repo, ours)
+    assert (now - start) < timedelta(seconds=5), "no room behind a send"
+    assert end == sent[1], "forward to the next send, not to the next received message"
+
+    # A received message keeps a centred window, reaching halfway to the received messages
+    # either side of it — and is untouched by our own sends in between.
+    theirs = ChatMessage(
+        text="theirs", outbound=False, peer="d4e5f6a7", created_at=now + timedelta(seconds=50)
+    )
+    start, end = direct_window(repo, theirs)
+    assert start == theirs.created_at - timedelta(seconds=40), "halfway back to the last one"
+    assert end == theirs.created_at + _DIRECT_WINDOW, "nothing follows, so that side stands"
+    repo.close()
+
+
+
+def test_collapse_folds_a_repeated_path_into_one_counted_row(tmp_path: Path) -> None:
+    """One row per path, carrying its copy count, first sighting and best SNR."""
+    from meshterm.services.message_paths import Arrival
+
+    base = utcnow()
+    arrivals = [
+        Arrival(when=base, hops=("3d63",), snr=10.0),
+        Arrival(when=base + timedelta(seconds=1), hops=("27d4",), snr=5.0),
+        Arrival(when=base + timedelta(seconds=5), hops=("3d63",), snr=13.5),
+    ]
+    folded = collapse(arrivals)
+    assert [a.hops for a in folded] == [("3d63",), ("27d4",)], "first-heard order"
+    assert folded[0].copies == 2 and folded[1].copies == 1
+    assert folded[0].when == base, "the first sighting is when the path first worked"
+    assert folded[0].snr == 13.5, "the best reading is what the path can do"
+