fix: TG 4000 clears STATUS and bridge legs after dynamic reset

Clear RX slot state to stop RPTO re-seeding cleared sessions, run
in-band 4000 deactivation on inject-only paths, and mark downlink dirty.

src/adn_server/application/routing/helpers.py

@@ -403,11 +403,14 @@ def seed_peer_ua_session_from_status(
 def clear_peer_rx_status_slots(
     status: dict[Any, Any],
     peer_id: bytes,
+    *,
+    slot: int | None = None,
 ) -> None:
     """Reset RX fields on slots last owned by this peer (avoids stale OPTIONS seed)."""
     pk = bytes_4(int_id(peer_id))
-    for slot in (1, 2):
-        slot_st = status.get(slot)
+    slots = (int(slot),) if slot is not None else (1, 2)
+    for slot_id in slots:
+        slot_st = status.get(slot_id)
         if not isinstance(slot_st, dict):
             continue
         if bytes_4(int_id(slot_st.get("RX_PEER", b"\x00"))) != pk:

src/adn_server/infrastructure/twisted_adapters/udp_hbp.py

@@ -59,7 +59,7 @@ from ...application.routing.peer_downlink_index import (
     invalidate_peer_options_cache,
 )
 from ...application.proxy.deployment import is_proxy_inject_only
-from ...domain import bytes_4, int_id
+from ...domain import bytes_3, bytes_4, int_id
 from ...domain.talker_alias import (
     DMRA_PACKET_LEN,
     decode_ta_from_blocks,
@@ -400,13 +400,17 @@ class HBPProtocol(DatagramProtocol):
         )
 
     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."""
+        """Clear per-peer UA dynamics and stale STATUS; deactivate bridges on 4000."""
         peer = self._peers.get(peer_id, {})
         clear_peer_ua_sessions(peer, self._config, peer_id, slot=slot)
+        clear_peer_rx_status_slots(self.STATUS, peer_id, slot=slot)
         if self._dynamic_tg_uc is not None:
             self._dynamic_tg_uc.delete_peer_slot(peer_id, self._system, slot)
+        if self._on_in_band_signalling:
+            self._on_in_band_signalling(self._system, slot, bytes_3(4000), time.time())
         _kind = "Private call to ID" if call_type == "unit" else "Group call to TG"
         if self._inject_multi_peer_options_filter():
+            self._mark_downlink_index_dirty()
             self._push_config_to_monitor()
             logger.info(
                 "(%s) %s 4000 received on TS %s — clearing dynamic TGs for peer %s",

tests/application/test_peer_single_downlink.py

@@ -147,6 +147,23 @@ def test_tg4000_clears_single_session() -> None:
     )
 
 
+def test_tg4000_clear_rx_status_blocks_rpto_reseed() -> None:
+    """After 4000, stale STATUS must not re-seed UA session on RPTO."""
+    peer = {"OPTIONS": b"TS2=730,7305;SINGLE=1;TIMER=5;", "CONNECTED": 999_000.0}
+    peer_id = _peer_id()
+    sys_cfg = _sys_cfg()
+    status = {2: {"RX_PEER": peer_id, "RX_TGID": bytes_3(7305), "RX_TIME": 999_500.0}}
+    now = 1_000_000.0
+    register_peer_ua_session(peer, peer_id, 2, 7305, sys_cfg, now=now)
+    clear_peer_ua_sessions(peer, sys_cfg, peer_id, slot=2)
+    clear_peer_rx_status_slots(status, peer_id, slot=2)
+    seed_peer_ua_session_from_status(peer, peer_id, 2, status[2], sys_cfg, now=now + 10)
+
+    assert peer_should_receive_group_voice(
+        peer, 2, 730, peer_id=peer_id, connected_count=8, sys_cfg=sys_cfg, now=now + 20
+    )
+
+
 def test_seed_session_from_status_after_options() -> None:
     """TX on 7305 before RPTO SINGLE=1 — OPTIONS must seed the lock."""
     peer = {"OPTIONS": b"TS2=730,7305;SINGLE=1;TIMER=5;", "CONNECTED": 999_000.0}