fix(neighbours): only treat a reception as our echo while that send is on air

The self-echo guard registered a packet's hash and forwarded path when the
engine decided to forward it and kept it for 30 s. Two things could then
hide a real neighbour from the link tracker:

- A neighbour whose 1-byte path hash collides with ours, repeating the
  same packet from the same upstream path, carries exactly the path we
  would have sent.
- The registration happened before the duty-cycle gate, so a packet that
  was never transmitted still suppressed matching receptions.

Registration now happens at the physical send, inside the TX lock after
the duty-cycle gate, and records which radio is sending. A send that
fails is forgotten. A reception counts as our echo only when it arrives on
a radio other than the one that sent it, and while that send is on the
air or within 2 s of it finishing, which covers a remote modem's delivery
delay. Single-radio nodes keep no records at all, since a lone radio
cannot hear itself.

A colliding neighbour is now mistaken for our echo only if it repeats the
same packet on the same path, on the other radio, inside that window.

repeater/engine.py

@@ -208,9 +208,9 @@ class RepeaterHandler(BaseHandler):
             config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF)
         )
         self.neighbour_link_tracker = NeighbourLinkTracker(config)
-        # (packet hash, path) of recent transmissions. Two radios sharing one
-        # frequency hear each other send; that echo ends in our own path hash and
-        # must not make this node its own neighbour.
+        # (packet hash, path) -> recent physical sends, each with its radio and
+        # timing. Two radios sharing one frequency hear each other send; that echo
+        # ends in our own path hash and must not make this node its own neighbour.
         self._recent_own_tx: OrderedDict = OrderedDict()
 
         radio = dispatcher.radio if dispatcher else None
@@ -405,7 +405,7 @@ class RepeaterHandler(BaseHandler):
                 self.radio_config["coding_rate"],
                 self.radio_config["preamble_length"],
             ).score
-            if not self._is_own_echo(pkt_hash_full, original_path_hashes):
+            if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id):
                 self.neighbour_link_tracker.observe(
                     packet,
                     route_type=route_type,
@@ -443,9 +443,6 @@ class RepeaterHandler(BaseHandler):
 
             # Capture the forwarded path (after modification)
             forwarded_path_hashes = fwd_pkt.get_path_hashes_hex()
-            # Before the first send: an echo can arrive while a later egress of
-            # the same fan-out is still on the air.
-            self._note_own_transmission(pkt_hash_full, forwarded_path_hashes)
 
             # MeshCore queues multi-ack redundancy copies ahead of the primary
             # ACK at the same scheduled time, so create their TX tasks first.
@@ -749,34 +746,83 @@ class RepeaterHandler(BaseHandler):
             return
         self._append_recent_packet(packet_record)
 
-    OWN_ECHO_TTL_SECONDS = 30.0
+    # Another radio of this node hears a send only while it is on the air or
+    # shortly after, allowing for a remote modem's delivery delay.
+    OWN_ECHO_MARGIN_SECONDS = 2.0
+    # A send that never reports back stops counting after this long.
+    OWN_ECHO_INFLIGHT_LIMIT_SECONDS = 60.0
     OWN_ECHO_MAX_ENTRIES = 256
 
-    def _note_own_transmission(self, packet_hash: str, path_hashes) -> None:
-        """Remember a packet about to be sent, so its echo can be recognised."""
+    def _note_own_send_started(self, packet: Packet, radio_id: Optional[str]):
+        """Record a physical send about to go on air; return a handle for its outcome.
+
+        Only on a node with two or more radios: a lone radio cannot hear itself.
+        """
+        _, radio_ids = self._fabric_endpoints()
+        if not radio_ids or len(radio_ids) < 2:
+            return None
+        path_hashes = packet.get_path_hashes_hex()
         if not path_hashes:
-            return
-        key = (packet_hash, tuple(path_hashes))
-        self._recent_own_tx.pop(key, None)
-        self._recent_own_tx[key] = time.monotonic()
+            return None
+        key = (packet.calculate_packet_hash().hex().upper(), tuple(path_hashes))
+        record = {"radio_id": radio_id, "started": time.monotonic(), "finished": None}
+        sends = self._recent_own_tx.pop(key, [])
+        sends.append(record)
+        self._recent_own_tx[key] = sends
         while len(self._recent_own_tx) > self.OWN_ECHO_MAX_ENTRIES:
             self._recent_own_tx.popitem(last=False)
+        return key, record
 
-    def _is_own_echo(self, packet_hash: str, path_hashes) -> bool:
-        """True when a reception carries exactly the path this node just sent.
+    def _note_own_send_finished(self, handle, packet: Packet, *, sent: bool) -> None:
+        """Close a send: keep it for its echo window, or forget it if nothing went on air."""
+        if handle is None:
+            return
+        key, record = handle
+        if record["finished"] is not None:
+            return
+        record["finished"] = time.monotonic()
+        if not sent:
+            sends = [r for r in self._recent_own_tx.get(key, []) if r is not record]
+            if sends:
+                self._recent_own_tx[key] = sends
+            else:
+                self._recent_own_tx.pop(key, None)
+            return
+        if record["radio_id"] is None:
+            metadata = getattr(packet, "_tx_metadata", None)
+            if isinstance(metadata, dict):
+                record["radio_id"] = metadata.get("radio_id") or metadata.get("tx_radio_id")
 
-        A neighbour repeating our copy appends its own hash, so only our own
-        transmission, heard by a second radio on the same frequency, matches.
+    def _own_send_is_live(self, record: dict, now: float) -> bool:
+        if record["finished"] is None:
+            return now - record["started"] <= self.OWN_ECHO_INFLIGHT_LIMIT_SECONDS
+        return now - record["finished"] <= self.OWN_ECHO_MARGIN_SECONDS
+
+    def _is_own_echo(self, packet_hash: str, path_hashes, rx_radio_id) -> bool:
+        """True when another radio of this node is hearing our own transmission.
+
+        The reception must carry exactly the path we sent, arrive on a radio other
+        than the one that sent it, and arrive while that send is on the air or
+        within OWN_ECHO_MARGIN_SECONDS of it finishing. A neighbour repeating our
+        copy appends its own hash; one whose 1-byte hash collides with ours is only
+        mistaken for an echo if it repeats the same packet on the same path inside
+        that window.
         """
-        if not path_hashes or not self._recent_own_tx:
+        if rx_radio_id is None or not path_hashes or not self._recent_own_tx:
             return False
         now = time.monotonic()
         while self._recent_own_tx:
-            oldest = next(iter(self._recent_own_tx.values()))
-            if now - oldest <= self.OWN_ECHO_TTL_SECONDS:
+            sends = next(iter(self._recent_own_tx.values()))
+            if any(self._own_send_is_live(record, now) for record in sends):
                 break
             self._recent_own_tx.popitem(last=False)
-        return (packet_hash, tuple(path_hashes)) in self._recent_own_tx
+        for record in self._recent_own_tx.get((packet_hash, tuple(path_hashes)), ()):
+            if not self._own_send_is_live(record, now):
+                continue
+            sender = record["radio_id"]
+            if sender is None or str(sender) != str(rx_radio_id):
+                return True
+        return False
 
     def record_duplicate(self, packet: Packet, rssi: int = 0, snr: float = 0.0) -> None:
         """Record a known-duplicate packet for UI/storage visibility without forwarding.
@@ -812,7 +858,8 @@ class RepeaterHandler(BaseHandler):
             self.radio_config["coding_rate"],
             self.radio_config["preamble_length"],
         ).score
-        if not self._is_own_echo(pkt_hash_full, original_path_hashes):
+        rx_radio_id = getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None)
+        if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id):
             self.neighbour_link_tracker.observe(
                 packet,
                 route_type=route_type,
@@ -821,9 +868,7 @@ class RepeaterHandler(BaseHandler):
                 snr=float(snr),
                 score=score,
                 is_duplicate=True,
-                rx_radio_id=(
-                    getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None)
-                ),
+                rx_radio_id=rx_radio_id,
             )
 
         packet_record = self._build_packet_record(
@@ -1780,6 +1825,9 @@ class RepeaterHandler(BaseHandler):
                             )
                             return False
 
+                    # Registered as the send starts: another radio of this node can
+                    # hear it while it is still on the air.
+                    own_send = self._note_own_send_started(fwd_pkt, preferred_tx_radio_id)
                     try:
                         if preferred_tx_radio_id:
                             try:
@@ -1794,6 +1842,7 @@ class RepeaterHandler(BaseHandler):
                                 )
                         else:
                             sent = await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False)
+                        self._note_own_send_finished(own_send, fwd_pkt, sent=bool(sent))
                         if not sent:
                             logger.warning(
                                 "Retransmit failed (attempt %d): dispatcher returned false",
@@ -1812,6 +1861,7 @@ class RepeaterHandler(BaseHandler):
                         )
                         return True
                     except Exception as e:
+                        self._note_own_send_finished(own_send, fwd_pkt, sent=False)
                         logger.error(f"Retransmit failed (attempt {attempt + 1}): {e}")
                         if local_transmission and attempt == 0:
                             pass  # release lock, outer loop sleeps, then retries

tests/test_fabric_fanout.py

@@ -289,6 +289,71 @@ async def test_own_transmission_heard_on_the_other_radio_is_not_a_neighbour():
     assert [r["drop_reason"] for r in rig.records()[1:]] == ["Duplicate"] * 3
 
 
+@pytest.mark.asyncio
+async def test_a_copy_heard_on_the_radio_that_sent_it_is_not_our_echo():
+    """A radio cannot hear its own transmission, so that copy came from a neighbour."""
+    rig = _Rig()  # bridge only: heard on local, sent on link
+    payload = b"\x51\x52\x53"
+    assert await rig.handler(_flood(payload=payload, path=b"\x11", rx="local"), _rx("local"))
+    assert rig.air.order == ["link"]
+
+    collided = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="link")
+    assert await rig.handler(collided, _rx("link")) is False
+    assert f"1:{LOCAL_HASH:02X}" in rig.handler.neighbour_link_tracker.links
+
+    echo = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="local")
+    rig.handler.neighbour_link_tracker.links.clear()
+    assert await rig.handler(echo, _rx("local")) is False
+    assert f"1:{LOCAL_HASH:02X}" not in rig.handler.neighbour_link_tracker.links
+
+
+@pytest.mark.asyncio
+async def test_a_matching_copy_after_the_echo_window_is_a_neighbour():
+    rig = _Rig({"repeat_on_ingress": True})
+    rig.handler.OWN_ECHO_MARGIN_SECONDS = -1.0  # every finished send is already outside it
+    payload = b"\x61\x62\x63"
+    assert await rig.handler(_flood(payload=payload, path=b"\x11", rx="local"), _rx("local"))
+
+    late = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="link")
+    assert await rig.handler(late, _rx("link")) is False
+
+    assert f"1:{LOCAL_HASH:02X}" in rig.handler.neighbour_link_tracker.links
+    assert not rig.handler._recent_own_tx  # expired sends are dropped
+
+
+@pytest.mark.asyncio
+async def test_a_send_refused_by_duty_cycle_leaves_no_echo_to_match():
+    rig = _Rig({"repeat_on_ingress": True})
+    rig.handler.airtime_mgr.can_transmit = MagicMock(return_value=(False, 5.0))
+    payload = b"\x71\x72\x73"
+    assert (
+        await rig.handler(_flood(payload=payload, path=b"\x11", rx="local"), _rx("local")) is False
+    )
+    assert rig.air.frames == []
+
+    collided = _flood(payload=payload, path=bytes([0x11, LOCAL_HASH]), rx="link")
+    assert await rig.handler(collided, _rx("link")) is False
+
+    assert not rig.handler._recent_own_tx
+    assert f"1:{LOCAL_HASH:02X}" in rig.handler.neighbour_link_tracker.links
+
+
+@pytest.mark.asyncio
+async def test_a_failed_send_is_forgotten():
+    rig = _Rig()
+    rig.link.fail = True
+    assert await rig.handler(_flood(path=b"\x11", rx="local"), _rx("local")) is False
+    assert not rig.handler._recent_own_tx
+
+
+@pytest.mark.asyncio
+async def test_a_single_radio_node_keeps_no_echo_records():
+    rig = _Rig()
+    with patch.object(rig.handler, "_fabric_endpoints", return_value=(None, ["local"])):
+        assert await rig.handler(_flood(path=b"\x11", rx="local"), _rx("local"))
+    assert not rig.handler._recent_own_tx
+
+
 @pytest.mark.asyncio
 async def test_egress_follows_captured_ingress_not_the_latest_rx():
     rig = _Rig({"repeat_on_ingress": True})

tests/test_tx_lock.py

@@ -17,6 +17,7 @@ or:
 import asyncio
 import time
 import unittest
+from collections import OrderedDict
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
@@ -61,6 +62,7 @@ def _make_handler():
     h.airtime_mgr.can_transmit.return_value = (True, 0.0)
     h.airtime_mgr.calculate_airtime.return_value = 100.0
     h._tx_lock = asyncio.Lock()
+    h._recent_own_tx = OrderedDict()
     h.sent_flood_count = 0
     h.sent_direct_count = 0
     # Stub out _record_packet_sent so it doesn't touch packet.header constants