fix: do not block downlink voice after SINGLE listen-lock VHEAD

Listen locks (source=listen) must not trip peer_single_blocks_foreign_same_tg_downlink;
same-stream contention stays in peer_hotspot_voice_slot_busy.

src/adn_server/application/routing/helpers.py

@@ -1331,7 +1331,12 @@ def peer_single_blocks_foreign_same_tg_downlink(
     *,
     now: float | None = None,
 ) -> bool:
-    """SINGLE=1 local UA on TG T: block network downlink on T unless hotspot is TX on T."""
+    """SINGLE=1 local UA on TG T: block network downlink on T unless hotspot is TX on T.
+
+    Downlink listen locks (``source=listen``) only exclude other TGs via
+    ``peer_single_blocks_group_voice``; same-TG stream overlap uses
+    ``peer_hotspot_voice_slot_busy``.
+    """
     if not sys_cfg or not peer_single_mode(peer, sys_cfg):
         return False
     incoming = int_id(incoming_tgid_b)
@@ -1340,6 +1345,9 @@ def peer_single_blocks_foreign_same_tg_downlink(
     )
     if locked is None or int(locked) != incoming:
         return False
+    entry = _peer_ua_session_entry(sys_cfg, peer_id, voice_slot)
+    if isinstance(entry, dict) and entry.get("source") == "listen":
+        return False
     active = (peer_slots or {}).get(int(voice_slot))
     if isinstance(active, dict) and active.get("ingress"):
         if int(active.get("tgid", 0) or 0) == incoming:

tests/application/test_peer_single_downlink.py

@@ -446,11 +446,50 @@ def test_downlink_listen_registers_single_session() -> None:
         bytes([0x80 | (HBPF_DATA_SYNC << 4) | HBPF_SLT_VHEAD]), bytes_4(0x11111111),
     ] + [b"\x00"] * 33)
     track_peer_group_dmrd(ctx, peer_id, vhead_7305, peer, pkt_time=now)
+    assert peer_should_receive_group_voice(
+        peer, 2, 7305, peer_id=peer_id, connected_count=3, sys_cfg=sys_cfg, now=now + 1,
+    )
     assert not peer_should_receive_group_voice(
         peer, 2, 730, peer_id=peer_id, connected_count=3, sys_cfg=sys_cfg, now=now + 1,
     )
 
 
+def test_downlink_listen_allows_voice_frames_after_vhead() -> None:
+    """Listen lock on VHEAD must not block voice frames of the same downlink stream."""
+    from adn_server.application.routing.downlink import (
+        DownlinkContext,
+        peer_accepts_group_dmrd_packet,
+        track_peer_group_dmrd,
+    )
+    from adn_server.domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, bytes_3, bytes_4
+    from adn_server.domain.hbp_protocol import HBPF_VOICE
+
+    sys_cfg = {"SINGLE_MODE": False, "DEFAULT_UA_TIMER": 10, "MODE": "MASTER", "MAX_PEERS": 8}
+    config = {"PROXY": {"TARGET_SYSTEM": "MASTER-A"}, "SYSTEMS": {"MASTER-A": sys_cfg}}
+    peer_id = _peer_id()
+    peer = {"OPTIONS": b"TS2=730502;SINGLE=1;TIMER=60;"}
+    ctx = DownlinkContext(
+        config=config,
+        system_name="MASTER-A",
+        sys_cfg=sys_cfg,
+        peers={peer_id: peer},
+        status={1: {}, 2: {}},
+        connected_count=3,
+    )
+    now = 1_000_000.0
+    stream = bytes_4(0x11111111)
+    vhead = b"".join([
+        b"DMRD", b"\x00", bytes_3(100), bytes_3(730502), b"\x00\x00\x00\x00",
+        bytes([0x80 | (HBPF_DATA_SYNC << 4) | HBPF_SLT_VHEAD]), stream,
+    ] + [b"\x00"] * 33)
+    voice = b"".join([
+        b"DMRD", b"\x00", bytes_3(100), bytes_3(730502), b"\x00\x00\x00\x00",
+        bytes([0x80 | (HBPF_VOICE << 4)]), stream,
+    ] + [b"\x00"] * 33)
+    track_peer_group_dmrd(ctx, peer_id, vhead, peer, pkt_time=now)
+    assert peer_accepts_group_dmrd_packet(ctx, peer_id, peer, voice)
+
+
 def test_downlink_vterm_clears_listen_session_not_local_tx() -> None:
     """Listen lock from downlink ends on VTERM; local TX session survives."""
     from adn_server.application.routing.downlink import (