fix: deliver a hotspot's own group call back to its other slot when subscribed there too

Adds self-echo (a hotspot with the same TG on both slots, static or dynamic, hears its own TX on the other slot) and fixes four occurrences of the same slot-resolution bug that blocked or truncated it: peer_downlink_voice_slot always preferred an unambiguous static/SINGLE=1-locked slot over the caller's own wire slot, corrupting the busy-check and the parrot/echo anti-loopback guard alike. Also makes the "Downlink dropped" log reason specific instead of generic, and throttles repeated identical drop logs.

src/adn_server/application/routing/downlink.py

@@ -34,7 +34,7 @@ from .helpers import (
     _peer_transmit_hangtime_blocks,
     _peer_ua_session_entry,
     clear_peer_ua_sessions,
-    hbp_slot_blocks_group_voice_for_peer,
+    hbp_slot_blocks_group_voice_for_peer_reason,
     is_special_tg,
     is_ua_session_tgid,
     master_per_peer_slot_contention,
@@ -159,12 +159,27 @@ def peer_hangtime_voice_slots(
     *,
     peer_id: bytes | None = None,
 ) -> set[int]:
-    """RF / OPTIONS slots that share transmit hangtime for this downlink."""
-    rf_slot = normalize_ua_voice_slot(peer, int(wire_slot))
+    """RF / OPTIONS slots that share transmit hangtime for this downlink.
+
+    ``wire_slot`` is already the caller's final, decided delivery slot (the
+    packet has already been remapped). If this peer is genuinely eligible for
+    ``tgid`` on ``wire_slot`` (static or dynamic, per the same combined logic
+    ``iter_downlink_voice_slots`` uses for the actual delivery decision),
+    trust it and check only that slot -- falling back to also deriving
+    ``rf_slot``/``listen_slot`` would pull in an unrelated slot (e.g. a static
+    slot for the same TG on this peer) whenever ``peer_downlink_voice_slot``'s
+    single-slot answer differs from ``wire_slot``, wrongly reporting that
+    unrelated slot's own busy/ingress state instead of ``wire_slot``'s.
+    """
+    wire_slot_i = int(wire_slot)
+    eligible = iter_downlink_voice_slots(peer, wire_slot_i, int(tgid), sys_cfg, peer_id=peer_id)
+    if wire_slot_i in eligible:
+        return {wire_slot_i}
+    rf_slot = normalize_ua_voice_slot(peer, wire_slot_i)
     listen_slot = peer_downlink_voice_slot(
-        peer, int(wire_slot), int(tgid), sys_cfg, peer_id=peer_id,
+        peer, wire_slot_i, int(tgid), sys_cfg, peer_id=peer_id,
     )
-    return {int(rf_slot), int(wire_slot), int(listen_slot)}
+    return {int(rf_slot), wire_slot_i, int(listen_slot)}
 
 
 def peer_slot_block_reason(
@@ -256,7 +271,7 @@ def peer_slot_block_reason(
     for voice_slot in sorted(voice_slots):
         hang_row = ctx.peer_voice_hangtime.get(pk, {}).get(voice_slot)
         slot_st = ctx.status.get(voice_slot, {})
-        if hbp_slot_blocks_group_voice_for_peer(
+        busy_reason = hbp_slot_blocks_group_voice_for_peer_reason(
             slot_st,
             peer_id,
             incoming_tgid_b,
@@ -269,8 +284,9 @@ def peer_slot_block_reason(
             peer_hang_row=hang_row,
             voice_slot=voice_slot,
             sys_cfg=ctx.sys_cfg,
-        ):
-            return f"slot {voice_slot} busy or in GROUP_HANGTIME"
+        )
+        if busy_reason is not None:
+            return busy_reason
     return None
 
 

src/adn_server/application/routing/helpers.py

@@ -336,7 +336,7 @@ def peer_single_same_tg_foreign_tx_blocks(
     return True
 
 
-def peer_hotspot_voice_slot_busy(
+def peer_hotspot_voice_slot_busy_reason(
     peer_id: bytes,
     voice_slot: int,
     stream_id: bytes,
@@ -350,8 +350,12 @@ def peer_hotspot_voice_slot_busy(
     peers: dict[Any, Any] | None = None,
     peer: dict[str, Any] | None = None,
     sys_cfg: dict[str, Any] | None = None,
-) -> bool:
-    """True when this hotspot must not receive another group stream on ``voice_slot``.
+) -> str | None:
+    """Reason this hotspot must not receive another group stream on ``voice_slot``, or None.
+
+    Same logic as ``peer_hotspot_voice_slot_busy`` (which just checks ``is not
+    None``) -- kept as one function so the diagnostic reason can never drift
+    from the actual accept/reject decision.
 
     Hard rules (SINGLE=0 and SINGLE=1):
 
@@ -366,10 +370,10 @@ def peer_hotspot_voice_slot_busy(
     if peer is None and peers is not None:
         peer = peers.get(peer_id) or peers.get(pk)
     if _peer_transmit_hangtime_blocks(hang_row, incoming_tgid_b, pkt_time, group_hangtime):
-        return True
+        return f"GROUP_HANGTIME: recent local transmit hanging on slot {voice_slot}"
+    incoming_tgid = int_id(incoming_tgid_b)
     active = (peer_slots or {}).get(int(voice_slot))
     if isinstance(active, dict):
-        incoming_tgid = int_id(incoming_tgid_b)
         active_tgid = int(active.get("tgid", 0) or 0)
         active_stream = active.get("stream_id")
         active_time = float(active.get("time", 0) or 0)
@@ -378,10 +382,10 @@ def peer_hotspot_voice_slot_busy(
             peer_slots.pop(int(voice_slot), None)
             active = None
         if isinstance(active, dict) and active.get("ingress"):
-            return True
+            return f"peer is transmitting (ingress) on slot {voice_slot}"
         if isinstance(active, dict) and active.get("bridge_hold") and active_tgid and incoming_tgid != active_tgid:
             if age <= group_hangtime:
-                return True
+                return f"bridge hold: TG {active_tgid} active on slot {voice_slot}"
             peer_slots.pop(int(voice_slot), None)
             active = None
         if isinstance(active, dict) and stream_id and active_stream:
@@ -391,24 +395,28 @@ def peer_hotspot_voice_slot_busy(
                 if age >= STREAM_TO:
                     peer_slots.pop(int(voice_slot), None)
                 else:
-                    return True
+                    return f"TG {incoming_tgid} already active on slot {voice_slot} with a different stream"
             elif active_tgid and incoming_tgid != active_tgid:
-                return True
+                return f"slot {voice_slot} busy with different TG {active_tgid}"
             else:
-                return True
+                return f"slot {voice_slot} busy with an unidentified active stream"
         elif isinstance(active, dict):
             if active_tgid and active_tgid == incoming_tgid:
                 pass
             else:
-                return True
+                return (
+                    f"slot {voice_slot} already occupied by TG {active_tgid}"
+                    if active_tgid
+                    else f"slot {voice_slot} already occupied by another stream"
+                )
     if isinstance(peer, dict) and peer_single_blocks_foreign_same_tg_downlink(
         peer, pk, voice_slot, incoming_tgid_b, peer_slots, sys_cfg, now=pkt_time,
     ):
-        return True
+        return f"SINGLE mode: local UA lock on TG {incoming_tgid} blocks foreign downlink on slot {voice_slot}"
     if isinstance(peer, dict) and peer_single_same_tg_foreign_tx_blocks(
         peer, pk, incoming_tgid_b, stream_id, slot_st, sys_cfg, pkt_time=pkt_time,
     ):
-        return True
+        return f"SINGLE mode: peer is transmitting a different call on TG {incoming_tgid}"
     if bytes_4(int_id(slot_st.get("RX_PEER", b""))) == pk:
         rx_active = (
             slot_st.get("RX_TYPE") is not None
@@ -416,12 +424,47 @@ def peer_hotspot_voice_slot_busy(
             and (pkt_time - float(slot_st.get("RX_TIME", 0))) < STREAM_TO
         )
         if rx_active and stream_id != slot_st.get("RX_STREAM_ID"):
-            return True
+            return f"slot {voice_slot} STATUS RX owner with a different active stream"
         if _peer_status_rx_hangtime_blocks(
             peer_id, slot_st, incoming_tgid_b, pkt_time, group_hangtime,
         ):
-            return True
-    return False
+            return f"GROUP_HANGTIME: recent STATUS RX hangtime on slot {voice_slot}"
+    return None
+
+
+def peer_hotspot_voice_slot_busy(
+    peer_id: bytes,
+    voice_slot: int,
+    stream_id: bytes,
+    incoming_tgid_b: bytes,
+    slot_st: dict[str, Any],
+    peer_slots: PeerVoiceSlotMap | None,
+    hang_row: tuple[int, float] | None,
+    pkt_time: float,
+    group_hangtime: float,
+    *,
+    peers: dict[Any, Any] | None = None,
+    peer: dict[str, Any] | None = None,
+    sys_cfg: dict[str, Any] | None = None,
+) -> bool:
+    """True when this hotspot must not receive another group stream on ``voice_slot``."""
+    return (
+        peer_hotspot_voice_slot_busy_reason(
+            peer_id,
+            voice_slot,
+            stream_id,
+            incoming_tgid_b,
+            slot_st,
+            peer_slots,
+            hang_row,
+            pkt_time,
+            group_hangtime,
+            peers=peers,
+            peer=peer,
+            sys_cfg=sys_cfg,
+        )
+        is not None
+    )
 
 
 def master_per_peer_slot_contention(
@@ -602,7 +645,7 @@ def obp_clear_deferred_bridge_tx_leg(
             obp_clear_flat_bridge_tx(slot_st)
 
 
-def hbp_slot_blocks_group_voice_for_peer(
+def hbp_slot_blocks_group_voice_for_peer_reason(
     slot_st: dict[str, Any],
     peer_id: bytes,
     incoming_tgid_b: bytes,
@@ -616,24 +659,21 @@ def hbp_slot_blocks_group_voice_for_peer(
     peer_hang_row: tuple[int, float] | None = None,
     voice_slot: int | None = None,
     sys_cfg: dict[str, Any] | None = None,
-) -> bool:
-    """Slot contention scoped to one hotspot when ``per_peer`` (inject-only multi-HS).
-
-    ``STATUS[slot]`` is shared at the MASTER, but each connected hotspot has an
-    independent RF timeslot. Another peer's active QSO must not block this peer.
+) -> str | None:
+    """Reason for ``hbp_slot_blocks_group_voice_for_peer``'s block, or None.
 
-    Inject-only OBP→HBP defers global contention and stamps bridge ``TX_*`` on the
-    shared slot row before ``send_peer``. Same-stream exemption must not treat that
-    bridge TX stamp as the hotspot's own leg while the peer is still on the air (RX).
+    Same logic as ``hbp_slot_blocks_group_voice_for_peer`` (which just checks
+    ``is not None``) -- kept as one function so the diagnostic reason can
+    never drift from the actual accept/reject decision.
     """
     if per_peer:
         if voice_slot is None:
-            return False
+            return None
         peer = None
         if peers is not None:
             pk = bytes_4(int_id(peer_id))
             peer = peers.get(peer_id) or peers.get(pk)
-        return peer_hotspot_voice_slot_busy(
+        return peer_hotspot_voice_slot_busy_reason(
             peer_id,
             int(voice_slot),
             stream_id,
@@ -647,8 +687,53 @@ def hbp_slot_blocks_group_voice_for_peer(
             peer=peer if isinstance(peer, dict) else None,
             sys_cfg=sys_cfg,
         )
-    return hbp_slot_blocks_group_voice(
+    if hbp_slot_blocks_group_voice(
         slot_st, incoming_tgid_b, stream_id, pkt_time, group_hangtime,
+    ):
+        return "global STATUS slot contention"
+    return None
+
+
+def hbp_slot_blocks_group_voice_for_peer(
+    slot_st: dict[str, Any],
+    peer_id: bytes,
+    incoming_tgid_b: bytes,
+    stream_id: bytes,
+    pkt_time: float,
+    group_hangtime: float,
+    *,
+    per_peer: bool,
+    peers: dict[Any, Any] | None = None,
+    peer_slots: PeerVoiceSlotMap | None = None,
+    peer_hang_row: tuple[int, float] | None = None,
+    voice_slot: int | None = None,
+    sys_cfg: dict[str, Any] | None = None,
+) -> bool:
+    """Slot contention scoped to one hotspot when ``per_peer`` (inject-only multi-HS).
+
+    ``STATUS[slot]`` is shared at the MASTER, but each connected hotspot has an
+    independent RF timeslot. Another peer's active QSO must not block this peer.
+
+    Inject-only OBP→HBP defers global contention and stamps bridge ``TX_*`` on the
+    shared slot row before ``send_peer``. Same-stream exemption must not treat that
+    bridge TX stamp as the hotspot's own leg while the peer is still on the air (RX).
+    """
+    return (
+        hbp_slot_blocks_group_voice_for_peer_reason(
+            slot_st,
+            peer_id,
+            incoming_tgid_b,
+            stream_id,
+            pkt_time,
+            group_hangtime,
+            per_peer=per_peer,
+            peers=peers,
+            peer_slots=peer_slots,
+            peer_hang_row=peer_hang_row,
+            voice_slot=voice_slot,
+            sys_cfg=sys_cfg,
+        )
+        is not None
     )
 
 
@@ -1793,6 +1878,17 @@ def peer_downlink_voice_slot(
     static = peer_options_static_tg_slot(peer, tgid)
     if static is not None:
         return static
+    from adn_server.application.report.payloads import parse_peer_options_static
+
+    ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS"))
+    tg = str(tgid)
+    if tg in ts1 and tg in ts2:
+        # Static on both slots (peer_options_static_tg_slot returns None
+        # because that's ambiguous *as a single answer*, not because the TG
+        # is unresolved) -- an in-progress SINGLE=1 exclusive lock or SINGLE=0
+        # UA_MULTI entry from this peer's own current transmission must not
+        # override a config that already, unambiguously, permits wire_slot.
+        return int(wire_slot)
     if sys_cfg is not None and peer_id is not None:
         tgid_i = int(tgid)
         pk = bytes_4(int_id(peer_id))
@@ -1806,6 +1902,10 @@ def peer_downlink_voice_slot(
         if isinstance(store, dict):
             per_peer = store.get(pk)
             if isinstance(per_peer, dict):
+                wire_slot_i = int(wire_slot)
+                wire_slot_set = per_peer.get(wire_slot_i)
+                if isinstance(wire_slot_set, set) and tgid_i in wire_slot_set:
+                    return wire_slot_i
                 for voice_slot in (1, 2):
                     slot_set = per_peer.get(voice_slot)
                     if isinstance(slot_set, set) and tgid_i in slot_set:

src/adn_server/infrastructure/twisted_adapters/udp_hbp.py

@@ -257,7 +257,7 @@ class HBPProtocol(DatagramProtocol):
             self._connected_peer_count = 0
             self._peer_voice_slots: dict[bytes, dict[int, dict[str, Any]]] = {}
             self._peer_voice_hangtime: dict[bytes, dict[int, tuple[int, float]]] = {}
-            self._downlink_drop_logged: set[tuple[bytes, bytes]] = set()
+            self._downlink_drop_logged: dict[tuple[bytes, str], float] = {}
             self._config_push_delayed = None
             self._config_push_throttle = ConfigPushThrottle()
             self._refresh_connected_peer_count()
@@ -672,9 +672,15 @@ class HBPProtocol(DatagramProtocol):
             ctx = self._downlink_ctx()
             if ctx.per_peer_contention():
                 peer = self._peers[peer_id]
-                voice_slot = peer_downlink_voice_slot(
-                    peer, slot, tgid, self._config, peer_id=peer_id,
-                )
+                # `packet` is already in its final, delivery-decided form (self-echo
+                # calls this via send_peer with the packet pre-remapped to its own
+                # other slot) -- trust its own wire slot rather than re-deriving one
+                # via peer_downlink_voice_slot, which prefers an unambiguous static
+                # slot regardless of which slot this delivery is actually for. That
+                # would wrongly resolve back to the peer's own currently-transmitting
+                # static slot and block a self-echo delivery to a different, free
+                # dynamic slot.
+                voice_slot = normalize_ua_voice_slot(peer, slot)
                 pk = bytes_4(int_id(peer_id))
                 active = ctx.peer_voice_slots.get(pk, {}).get(int(voice_slot))
                 if isinstance(active, dict) and active.get("ingress"):
@@ -748,18 +754,11 @@ class HBPProtocol(DatagramProtocol):
             self._downlink_ctx(), peer_id, peer, packet, routed=routed,
         )
 
-    def _downlink_drop_key(self, peer_id: bytes, route_pkt: bytes) -> tuple[bytes, bytes]:
-        pk = bytes_4(int_id(peer_id))
-        stream_id = route_pkt[16:20] if len(route_pkt) >= 20 else b""
-        return pk, stream_id
+    _DOWNLINK_DROP_LOG_COOLDOWN = 5.0
 
     def _log_downlink_drop_once(
         self, peer_id: bytes, route_pkt: bytes, peer: dict[str, Any] | None,
     ) -> None:
-        key = self._downlink_drop_key(peer_id, route_pkt)
-        if key in self._downlink_drop_logged:
-            return
-        self._downlink_drop_logged.add(key)
         # By the time send_peer reaches this call, _peer_should_receive_dmrd
         # already passed (it returns silently, without logging, on its own
         # earlier check) -- so the only thing left that could have rejected
@@ -769,6 +768,17 @@ class HBPProtocol(DatagramProtocol):
         reason = "unknown"
         if isinstance(peer, dict):
             reason = peer_slot_block_reason(self._downlink_ctx(), peer_id, peer, route_pkt) or "unknown"
+        # Keyed on (peer, reason) rather than stream_id: a still-ongoing call
+        # keeps hitting the same reason on every voice frame, so per-stream
+        # dedup alone doesn't stop the spam. Cooldown re-logs periodically
+        # instead of only once, so a long-running block doesn't go silent.
+        pk = bytes_4(int_id(peer_id))
+        key = (pk, reason)
+        now = time.time()
+        last = self._downlink_drop_logged.get(key)
+        if last is not None and (now - last) < self._DOWNLINK_DROP_LOG_COOLDOWN:
+            return
+        self._downlink_drop_logged[key] = now
         logger.info(
             "(%s) Downlink dropped for peer %s TG %s (%s)",
             self._system,
@@ -797,7 +807,6 @@ class HBPProtocol(DatagramProtocol):
             ):
                 self._log_downlink_drop_once(_peer, route_pkt, peer)
                 return
-            self._downlink_drop_logged.discard(self._downlink_drop_key(_peer, route_pkt))
             _packet = b"".join([route_pkt[:11], _peer, route_pkt[15:]])
             ctx = self._downlink_ctx()
             if isinstance(peer, dict):
@@ -1415,6 +1424,26 @@ class HBPProtocol(DatagramProtocol):
                         for _peer in _pvt_targets:
                             if _peer != _peer_id:
                                 self.send_peer(_peer, _repeat_pkt)
+                    # Same repeater, other slot: if this peer is also subscribed
+                    # (static or dynamic) to this TG on the slot it did NOT just
+                    # transmit on, deliver there too -- that slot is a genuinely
+                    # independent RF path also tuned to this TG. Purely additive:
+                    # does not change delivery to any other peer above.
+                    if _call_type in ("group", "vcsbk"):
+                        _src_peer_obj = self._peers.get(_peer_id)
+                        if isinstance(_src_peer_obj, dict):
+                            _src_voice_slots = iter_downlink_voice_slots(
+                                _src_peer_obj, _slot, int_id(_dst_id),
+                                self._config, peer_id=_peer_id,
+                            )
+                            for _src_vs in _src_voice_slots:
+                                if _src_vs == _slot:
+                                    continue
+                                _echo_remapped = remap_dmrd_for_peer(
+                                    _repeat_pkt, _src_peer_obj, self._config,
+                                    peer_id=_peer_id, voice_slot=_src_vs,
+                                )
+                                self.send_peer(_peer_id, _echo_remapped, _skip_dual_expand=True)
                 # TG 4000: reset after REPEAT so peers see the packet (legacy order)
                 if self._handle_tg4000_packet(
                     _peer_id, _slot, _int_dst_id, _call_type, _frame_type, _dtype_vseq,

tests/application/test_peer_rf_mode.py

@@ -11,6 +11,7 @@ from adn_server.application.routing.helpers import (
     RF_MODE_DUPLEX,
     RF_MODE_SIMPLEX,
     SIMPLEX_VOICE_SLOT,
+    bytes_4,
     derive_peer_rf_mode,
     peer_downlink_voice_slot,
     peer_is_simplex,
@@ -84,6 +85,47 @@ def test_duplex_keeps_cross_slot_static_remap() -> None:
     assert not (remapped[15] & 0x80)
 
 
+def test_downlink_voice_slot_prefers_wire_slot_when_dynamic_tg_active_on_both() -> None:
+    # Regression: a SINGLE=0 (UA_MULTI) TG dynamically active on both slots
+    # used to always resolve to slot 1 regardless of which slot was asked
+    # about, wrongly reporting the peer's own actively-transmitting slot as
+    # the "listen slot" for the opposite-direction self-echo delivery.
+    peer = _duplex_peer()
+    peer_rf_mode(peer)
+    peer_id = 714001
+    pk = bytes_4(peer_id)
+    sys_cfg = {"_PEER_UA_MULTI_TGS": {pk: {1: {71442}, 2: {71442}}}}
+    assert peer_downlink_voice_slot(peer, 1, 71442, sys_cfg, peer_id=peer_id) == 1
+    assert peer_downlink_voice_slot(peer, 2, 71442, sys_cfg, peer_id=peer_id) == 2
+
+
+def test_downlink_voice_slot_static_both_slots_not_hijacked_by_single_exclusive_lock() -> None:
+    # Regression: a peer with a TG static on BOTH TS1 and TS2 (SINGLE=1) that
+    # is currently transmitting registers a SINGLE=1 exclusive session lock
+    # on its own TX slot. peer_downlink_voice_slot used to consult that lock
+    # (checking slot 1 before slot 2, unconditionally) whenever the static
+    # config was "ambiguous" (both slots), hijacking the answer to whichever
+    # slot the peer's own lock happened to be on -- wrongly reporting the
+    # peer's own actively-transmitting slot as busy for a self-echo delivery
+    # actually targeting the *other* slot.
+    peer = {
+        "SLOTS": b"3",
+        "RX_FREQ": b"431612500",
+        "TX_FREQ": b"438612500",
+        "OPTIONS": b"TS1=730500;TS2=730500;SINGLE=1;",
+    }
+    peer_rf_mode(peer)
+    peer_id = 730039258
+    pk = bytes_4(peer_id)
+    sys_cfg = {
+        "_PEER_UA_SESSIONS": {
+            pk: {1: {"tgid": 730500, "expires": 0, "source": "local"}},
+        },
+    }
+    assert peer_downlink_voice_slot(peer, 1, 730500, sys_cfg, peer_id=peer_id) == 1
+    assert peer_downlink_voice_slot(peer, 2, 730500, sys_cfg, peer_id=peer_id) == 2
+
+
 def test_topology_peer_row_includes_rf_mode() -> None:
     peer = _simplex_peer()
     peer_rf_mode(peer)

tests/application/test_slot_contention.py

@@ -11,7 +11,9 @@ from adn_server.application.routing.helpers import (
     hbp_master_ingress_repeat_allowed,
     hbp_slot_blocks_group_voice,
     hbp_slot_blocks_group_voice_for_peer,
+    hbp_slot_blocks_group_voice_for_peer_reason,
     peer_hotspot_voice_slot_busy,
+    peer_hotspot_voice_slot_busy_reason,
     register_peer_ua_session,
     slot_has_active_voice,
     slot_in_group_hangtime,
@@ -333,6 +335,54 @@ def test_peer_hotspot_voice_slot_busy_blocks_downlink_during_local_ingress() ->
     )
 
 
+def test_peer_hotspot_voice_slot_busy_reason_distinguishes_ingress_and_bridge_hold() -> None:
+    """Regression: the drop reason must name the actual cause, not a generic string."""
+    now = 1_000_000.0
+    hs = bytes_4(730039265)
+    slot = _active_rx_slot()
+    slot["RX_PEER"] = bytes_4(730039264)
+
+    ingress_slots = {
+        2: {"stream_id": _STREAM_B, "tgid": 730502, "time": now, "ingress": True},
+    }
+    reason = peer_hotspot_voice_slot_busy_reason(
+        hs, 2, _STREAM_A, _TG_A, slot, ingress_slots, None, now + 0.05, 5.0,
+    )
+    assert reason is not None and "transmitting (ingress) on slot 2" in reason
+    assert peer_hotspot_voice_slot_busy(
+        hs, 2, _STREAM_A, _TG_A, slot, ingress_slots, None, now + 0.05, 5.0,
+    ) == (reason is not None)
+
+    bridge_slots = {
+        2: {"stream_id": _STREAM_B, "tgid": 71442, "time": now, "bridge_hold": True},
+    }
+    reason = peer_hotspot_voice_slot_busy_reason(
+        hs, 2, _STREAM_A, _TG_A, slot, bridge_slots, None, now + 0.05, 5.0,
+    )
+    assert reason is not None and "bridge hold: TG 71442 active on slot 2" in reason
+    assert peer_hotspot_voice_slot_busy(
+        hs, 2, _STREAM_A, _TG_A, slot, bridge_slots, None, now + 0.05, 5.0,
+    ) == (reason is not None)
+
+
+def test_hbp_slot_blocks_group_voice_for_peer_reason_matches_bool() -> None:
+    now = 1_000_000.0
+    peer_a = bytes_4(352000133)
+    peer_b = bytes_4(714002301)
+    slot = _active_rx_slot()
+    slot["RX_PEER"] = peer_a
+    assert hbp_slot_blocks_group_voice_for_peer_reason(
+        slot, peer_b, _TG_B, _STREAM_B, now + 0.1, 0.0, per_peer=True, voice_slot=2,
+    ) is None
+    reason = hbp_slot_blocks_group_voice_for_peer_reason(
+        slot, peer_a, _TG_B, _STREAM_B, now + 0.1, 0.0, per_peer=True, voice_slot=2,
+    )
+    assert reason is not None
+    assert hbp_slot_blocks_group_voice_for_peer(
+        slot, peer_a, _TG_B, _STREAM_B, now + 0.1, 0.0, per_peer=True, voice_slot=2,
+    ) is True
+
+
 def test_per_peer_scope_ignores_other_hotspot_busy_slot() -> None:
     """Inject-only: peer B must not inherit slot contention from peer A."""
     now = 1_000_000.0

tests/infrastructure/test_hbp_repeat_group_dual_slot.py

@@ -152,6 +152,88 @@ def test_group_call_static_on_one_slot_dynamic_on_other_repeats_to_both() -> Non
     assert slots == [1, 2]
 
 
+def test_group_call_echoes_to_transmitting_peers_own_other_slot() -> None:
+    """New behavior: a repeater transmitting a TG on one slot, that is also
+    subscribed (static or dynamic) to that same TG on its OTHER slot, hears
+    its own call echoed there too -- that other slot is a genuinely
+    independent RF path also tuned to the TG."""
+    stack = build_hbp_repeat_stack(talker_alias=False)
+    stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False
+    # TG static on TS1 only; transmits on slot 2 -- slot 1 is the "other" slot.
+    stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS1={_TG};")
+
+    base = _group_spec(slot=2)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX)
+    stack.transport.clear()
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+
+    echo = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_TX and pkt[:4] == DMRD]
+    slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in echo)
+    assert slots == [1], f"peer should hear itself echoed on its own other slot -- got {slots}"
+
+
+def test_group_call_echoes_to_own_dynamically_subscribed_other_slot() -> None:
+    """Same self-echo, but the other slot's subscription is dynamic (SINGLE=0
+    keyed), not static -- static vs dynamic must be treated identically."""
+    stack = build_hbp_repeat_stack(talker_alias=False)
+    stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False
+    stack.register_peer(_PEER_TX, _ADDR_TX, options="")
+    pk = bytes_4(int.from_bytes(_PEER_TX, "big"))
+    stack.config["SYSTEMS"][stack.system_name].setdefault("_PEER_UA_MULTI_TGS", {})[pk] = {
+        1: {_TG},
+    }
+
+    base = _group_spec(slot=2)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX)
+    stack.transport.clear()
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+
+    echo = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_TX and pkt[:4] == DMRD]
+    slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in echo)
+    assert slots == [1], f"peer should hear itself echoed on its dynamically-subscribed other slot -- got {slots}"
+
+
+def test_group_call_echoes_symmetrically_on_opposite_slot() -> None:
+    """Same behavior, transmitting on the opposite slot: TG static on TS2,
+    transmits on slot 1 -- echoes back to itself on slot 2."""
+    stack = build_hbp_repeat_stack(talker_alias=False)
+    stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False
+    stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS2={_TG};")
+
+    base = _group_spec(slot=1)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX)
+    stack.transport.clear()
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+
+    echo = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_TX and pkt[:4] == DMRD]
+    slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in echo)
+    assert slots == [2], f"peer should hear itself echoed on its own other slot -- got {slots}"
+
+
+def test_group_call_no_self_echo_when_not_subscribed_on_other_slot() -> None:
+    """No TG configured at all -- no self-echo, matching existing (unchanged)
+    single-peer behavior."""
+    stack = build_hbp_repeat_stack(talker_alias=False)
+    stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False
+    stack.register_peer(_PEER_TX, _ADDR_TX, options="")
+
+    base = _group_spec(slot=2)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX)
+    stack.transport.clear()
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+
+    echo = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_TX and pkt[:4] == DMRD]
+    assert echo == []
+
+
 def test_group_call_on_dynamic_tg_single_mode_stays_exclusive_to_one_slot() -> None:
     """SINGLE=1 ("one exclusive dynamic TG per hotspot, either RF slot; new
     local TX replaces all others" -- register_peer_ua_session) cannot have
@@ -178,3 +260,69 @@ def test_group_call_on_dynamic_tg_single_mode_stays_exclusive_to_one_slot() -> N
     downlink = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_RX and pkt[:4] == DMRD]
     slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in downlink)
     assert slots == [2]
+
+
+def test_group_call_self_echo_not_blocked_by_own_ingress_when_dynamic_tg_on_both_slots() -> None:
+    """Regression: when a TG is dynamically (SINGLE=0) active on both slots,
+    peer_downlink_voice_slot used to always resolve to slot 1 regardless of
+    which slot was asked about. That made the busy-check for a slot-1-to-2
+    self-echo also examine slot 1 -- the peer's own live ingress slot -- so
+    the echo was wrongly reported busy even though slot 2 was free."""
+    stack = build_hbp_repeat_stack(talker_alias=False)
+    stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False
+    stack.register_peer(_PEER_TX, _ADDR_TX, options="")
+    pk = bytes_4(int.from_bytes(_PEER_TX, "big"))
+    stack.config["SYSTEMS"][stack.system_name].setdefault("_PEER_UA_MULTI_TGS", {})[pk] = {
+        1: {_TG}, 2: {_TG},
+    }
+
+    base = _group_spec(slot=1)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base), _ADDR_TX)
+    stack.transport.clear()
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+
+    echo = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_TX and pkt[:4] == DMRD]
+    slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in echo)
+    assert slots == [2], (
+        "self-echo to slot 2 must not be blocked by the source's own ingress "
+        f"on slot 1 -- got {slots}"
+    )
+
+
+def test_group_call_self_echo_to_dynamic_slot_not_blocked_by_own_static_slot_ingress() -> None:
+    """Regression: TG static on TS1 only, dynamically activated on TS2 by an
+    earlier call. peer_hangtime_voice_slots used to always union in the
+    static slot (TS1) as a busy-check candidate, via peer_downlink_voice_slot,
+    even when the caller only cares about TS2 -- so a later TX on TS1 (the
+    static slot, now busy with the source's own ingress) wrongly blocked its
+    own self-echo delivery to TS2, even though TS2 was free."""
+    stack = build_hbp_repeat_stack(talker_alias=False)
+    stack.config["SYSTEMS"][stack.system_name]["SINGLE_MODE"] = False
+    stack.register_peer(_PEER_TX, _ADDR_TX, options=f"TS1={_TG};SINGLE=0;")
+
+    # First call: TX on TS2 (not in static OPTIONS) -- dynamically activates
+    # TS2, and (as a side effect) self-echoes back to the static TS1 slot.
+    base_a = _group_spec(slot=2)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base_a), _ADDR_TX)
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base_a, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+    stack.inject_spec(DeterministicScenario.voice_term_spec(base_a, seq=2), _ADDR_TX)
+
+    # Second call: TX on TS1 (the static slot) -- must self-echo to TS2,
+    # which is now dynamically active from the first call.
+    stack.transport.clear()
+    base_b = _group_spec(slot=1)
+    stack.inject_spec(DeterministicScenario.voice_head_spec(base_b), _ADDR_TX)
+    stack.inject_spec(
+        DeterministicScenario.voice_burst_spec(base_b, seq=1, dtype_vseq=1), _ADDR_TX,
+    )
+
+    echo = [pkt for pkt, addr in stack.transport.sent if addr == _ADDR_TX and pkt[:4] == DMRD]
+    slots = sorted(parse_dmr_fields(pkt)["slot"] for pkt in echo)
+    assert slots == [2, 2], (
+        "self-echo to the dynamically-active TS2 must not be blocked by the "
+        f"source's own busy static TS1 -- got {slots}"
+    )