fix(inject): TG 4000 clears dynamics once per PTT and per peer only

Stop deactivating on every voice burst and scope inject-only resets to the
transmitting hotspot instead of all system dynamic bridges.

src/adn_server/application/bridge/helpers.py

@@ -8,13 +8,22 @@ from __future__ import annotations
 import time
 from typing import Any
 
-from ...domain import bytes_3, bytes_4, int_id
+from ...domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, bytes_3, bytes_4, int_id
 
 # Embedded LC codeword sits at bits 116:148 inside the 48-bit EMB field (108:156).
 # Legacy bridge_master.py replaces dmrbits[116:148] on bursts B–E (dtype_vseq 1–4).
 EMB_LC_SLICE = slice(116, 148)
 
 
+def tg4000_reset_on_vhead(int_dst_id: int, frame_type: int, dtype_vseq: int) -> bool:
+    """True when TG/ID 4000 voice header should trigger a one-shot dynamic reset."""
+    return (
+        int_dst_id == 4000
+        and frame_type == HBPF_DATA_SYNC
+        and dtype_vseq == HBPF_SLT_VHEAD
+    )
+
+
 def obp_target_bcsq_quenches_stream(
     systems_cfg: dict[str, Any], target_name: str, dst_id_b: bytes, stream_id: bytes
 ) -> bool:

src/adn_server/infrastructure/twisted_adapters/udp_hbp.py

@@ -50,6 +50,7 @@ from ...application.bridge.helpers import (
     peer_single_exclusive_tgid,
     register_peer_ua_session,
     seed_peer_ua_session_from_status,
+    tg4000_reset_on_vhead,
 )
 from ...application.proxy.deployment import is_proxy_inject_only
 from ...domain import bytes_4, int_id
@@ -279,6 +280,41 @@ class HBPProtocol(DatagramProtocol):
         """Inject-only proxy: always filter downlink by each peer's own OPTIONS."""
         return is_proxy_inject_only(self._CONFIG, self._system)
 
+    def _apply_tg4000_reset(self, peer_id: bytes, slot: int, call_type: str) -> None:
+        """Clear per-peer UA dynamics; legacy bridge reset only outside inject-only."""
+        peer = self._peers.get(peer_id, {})
+        clear_peer_ua_sessions(peer, self._config, peer_id, slot=slot)
+        _kind = "Private call to ID" if call_type == "unit" else "Group call to TG"
+        if self._inject_multi_peer_options_filter():
+            self._push_config_to_monitor()
+            logger.info(
+                "(%s) %s 4000 received on TS %s — clearing dynamic TGs for peer %s",
+                self._system, _kind, slot, int_id(peer_id),
+            )
+            return
+        if self._on_deactivate_dynamic_bridges:
+            logger.info(
+                "(%s) %s 4000 received on TS %s — deactivating all dynamic bridges",
+                self._system, _kind, slot,
+            )
+            self._on_deactivate_dynamic_bridges(self._system)
+
+    def _handle_tg4000_packet(
+        self,
+        peer_id: bytes,
+        slot: int,
+        int_dst_id: int,
+        call_type: str,
+        frame_type: int,
+        dtype_vseq: int,
+    ) -> bool:
+        """Legacy early return for TG/ID 4000; reset once per PTT on voice header only."""
+        if int_dst_id != 4000:
+            return False
+        if tg4000_reset_on_vhead(int_dst_id, frame_type, dtype_vseq):
+            self._apply_tg4000_reset(peer_id, slot, call_type)
+        return True
+
     def _peer_should_receive_dmrd(self, peer_id: bytes, packet: bytes) -> bool:
         if not self._inject_multi_peer_options_filter():
             return True
@@ -780,11 +816,7 @@ class HBPProtocol(DatagramProtocol):
                         peer_id=_peer_id,
                         now=pkt_time,
                     )
-                    if _int_dst_id == 4000:
-                        clear_peer_ua_sessions(
-                            self._peers[_peer_id], self._config, _peer_id, slot=_slot,
-                        )
-                    else:
+                    if _int_dst_id != 4000:
                         register_peer_ua_session(
                             self._peers[_peer_id],
                             _peer_id,
@@ -793,8 +825,8 @@ class HBPProtocol(DatagramProtocol):
                             self._config,
                             now=pkt_time,
                         )
-                    if _int_dst_id != 4000 and _prev_single_tg != _int_dst_id:
-                        self._push_config_to_monitor()
+                        if _prev_single_tg != _int_dst_id:
+                            self._push_config_to_monitor()
                 if (
                     _call_type in ("group", "vcsbk")
                     and _frame_type != HBPF_DATA_SYNC
@@ -833,14 +865,10 @@ class HBPProtocol(DatagramProtocol):
                     for _peer in self._peers:
                         if _peer != _peer_id:
                             self.send_peer(_peer, _repeat_pkt)
-                # TG 4000: deactivate after REPEAT so peers see the packet (legacy order)
-                if _int_dst_id == 4000 and self._on_deactivate_dynamic_bridges:
-                    _kind = "Private call to ID" if _call_type == "unit" else "Group call to TG"
-                    logger.info(
-                        "(%s) %s 4000 received on TS %s — deactivating all dynamic bridges",
-                        self._system, _kind, _slot,
-                    )
-                    self._on_deactivate_dynamic_bridges(self._system)
+                # 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,
+                ):
                     return
                 if _call_type == "group" and _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD:
                     logger.info(
@@ -1220,14 +1248,10 @@ class HBPProtocol(DatagramProtocol):
                 sub_map = self._CONFIG.get("_SUB_MAP")
                 if sub_map is not None:
                     sub_map[_rf_src] = (self._system, _slot, pkt_time)
-                # TG 4000: deactivate after ACL/SUB_MAP (legacy order — routerHBP.dmrd_received)
-                if _int_dst_id == 4000 and self._on_deactivate_dynamic_bridges:
-                    _kind = "Private call to ID" if _call_type == "unit" else "Group call to TG"
-                    logger.info(
-                        "(%s) %s 4000 received on TS %s — deactivating all dynamic bridges",
-                        self._system, _kind, _slot,
-                    )
-                    self._on_deactivate_dynamic_bridges(self._system)
+                # TG 4000: reset after ACL/SUB_MAP (legacy order — routerHBP.dmrd_received)
+                if self._handle_tg4000_packet(
+                    _peer_id, _slot, _int_dst_id, _call_type, _frame_type, _dtype_vseq,
+                ):
                     return
                 if _call_type == "group" and _frame_type == HBPF_DATA_SYNC and _dtype_vseq == HBPF_SLT_VHEAD:
                     logger.info(

tests/application/test_peer_single_downlink.py

@@ -3,10 +3,11 @@
 from __future__ import annotations
 
 
-from adn_server.domain import bytes_3, bytes_4, int_id
+from adn_server.domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, bytes_3, bytes_4, int_id
 
 from adn_server.application.bridge.helpers import (
     clear_peer_ua_sessions,
+    tg4000_reset_on_vhead,
     clear_peer_rx_status_slots,
     peer_should_receive_group_voice,
     peer_single_blocks_group_voice,
@@ -97,6 +98,12 @@ def test_single_zero_allows_multiple_static_tgs() -> None:
     )
 
 
+def test_tg4000_reset_only_on_voice_header() -> None:
+    assert tg4000_reset_on_vhead(4000, HBPF_DATA_SYNC, HBPF_SLT_VHEAD)
+    assert not tg4000_reset_on_vhead(4000, HBPF_DATA_SYNC, 2)
+    assert not tg4000_reset_on_vhead(730, HBPF_DATA_SYNC, HBPF_SLT_VHEAD)
+
+
 def test_tg4000_clears_single_session() -> None:
     peer = {"OPTIONS": b"TS2=730,7305;SINGLE=1;TIMER=5;"}
     sys_cfg = _sys_cfg()