refactor(voice): address review on the prompt slot hold

- The state a prompt shares between its thread and the reactor is a
  small dataclass (_PromptRun) instead of a dict with string keys.
- "Is the slot held by another stream?" now lives in routing/helpers.py
  next to slot_has_active_voice, and both read the RX/TX legs through
  one helper, instead of a third copy of that check in voice_use_cases.
- Blank lines and the test header follow the rest of the repo.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/adn_server/application/routing/helpers.py

@@ -112,17 +112,24 @@ def peer_is_simplex(peer: dict[str, Any]) -> bool:
     return peer_rf_mode(peer) == RF_MODE_SIMPLEX
 
 
+def _slot_leg_active(slot_st: dict[str, Any], leg: str, pkt_time: float) -> bool:
+    """True when the slot's ``RX`` or ``TX`` leg carries voice (within STREAM_TO)."""
+    leg_type = slot_st.get(f"{leg}_TYPE")
+    if leg_type is None or leg_type == HBPF_SLT_VTERM:
+        return False
+    return (pkt_time - float(slot_st.get(f"{leg}_TIME", 0) or 0)) < STREAM_TO
+
+
 def slot_has_active_voice(slot_st: dict[str, Any], pkt_time: float) -> bool:
     """True when the slot has an active group-voice RX or TX leg (within STREAM_TO)."""
-    rx_type = slot_st.get("RX_TYPE")
-    if rx_type is not None and rx_type != HBPF_SLT_VTERM:
-        if (pkt_time - float(slot_st.get("RX_TIME", 0))) < STREAM_TO:
-            return True
-    tx_type = slot_st.get("TX_TYPE")
-    if tx_type is not None and tx_type != HBPF_SLT_VTERM:
-        if (pkt_time - float(slot_st.get("TX_TIME", 0))) < STREAM_TO:
-            return True
-    return False
+    return _slot_leg_active(slot_st, "RX", pkt_time) or _slot_leg_active(slot_st, "TX", pkt_time)
+
+
+def slot_voice_held_by_other_stream(slot_st: dict[str, Any], stream_id: bytes, pkt_time: float) -> bool:
+    """Like ``slot_has_active_voice``, but ``stream_id``'s own TX leg does not count."""
+    if _slot_leg_active(slot_st, "RX", pkt_time):
+        return True
+    return slot_st.get("TX_STREAM_ID") != stream_id and _slot_leg_active(slot_st, "TX", pkt_time)
 
 
 def _slot_last_voice_activity(slot_st: dict[str, Any]) -> tuple[bytes, float]:

src/adn_server/application/voice_use_cases.py

@@ -28,12 +28,13 @@ from __future__ import annotations
 
 import logging
 import time
+from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Callable
 
 from ..domain import HBPF_SLT_VHEAD, HBPF_SLT_VTERM, bytes_3, bytes_4, int_id
-from ..domain.hbp_protocol import STREAM_TO
 from .ports import VoiceProvider
+from .routing.helpers import slot_voice_held_by_other_stream
 from .server_voice import (
     announcement_item_source_bytes,
     server_voice_rf_src_bytes,
@@ -46,19 +47,17 @@ _ANNOUNCEMENT_EXCLUDED = ("ECHO", "D-APRS")
 _BROADCAST_GAP = 1.5
 
 
+@dataclass
+class _PromptRun:
+    """State of one prompt, shared by its worker thread and the reactor.
+
+    Only the reactor writes it; the thread reads ``stopped`` between frames.
+    """
+
+    stopped: bool = False
+    sent: int = 0
+    stream_id: bytes | None = None
 
-def _slot_taken_by_other_stream(slot: dict[str, Any], stream_id: bytes, now: float) -> bool:
-    """A radio is talking on the slot, or another stream is being sent to it."""
-    rx_type = slot.get("RX_TYPE")
-    if rx_type is not None and rx_type != HBPF_SLT_VTERM and now - float(slot.get("RX_TIME", 0) or 0) < STREAM_TO:
-        return True
-    tx_type = slot.get("TX_TYPE")
-    return (
-        tx_type is not None
-        and tx_type != HBPF_SLT_VTERM
-        and slot.get("TX_STREAM_ID") != stream_id
-        and now - float(slot.get("TX_TIME", 0) or 0) < STREAM_TO
-    )
 
 class VoiceUseCases:
     """Use cases for voice announcements and TTS."""
@@ -915,10 +914,10 @@ class VoiceUseCases:
         as a second stream on the same slot. A radio keying up, or a call already
         on the slot, stops the prompt; the slot is released when it ends.
         """
-        run: dict[str, Any] = {"stopped": False, "sent": 0, "stream_id": None}
+        run = _PromptRun()
         _next_time = time.time()
         for pkt in speech:
-            if run["stopped"]:
+            if run.stopped:
                 break
             _next_time += _FRAME_INTERVAL
             delay = _next_time - time.time()
@@ -926,38 +925,36 @@ class VoiceUseCases:
                 time.sleep(delay)
             self._call_from_reactor(self._prompt_frame, protocol, system, pkt, source_id, dst_id, run)
         self._call_from_reactor(self._prompt_end, protocol, run)
-        return int(run["sent"])
+        return run.sent
 
     def _prompt_frame(
-        self, protocol: Any, system: str, pkt: bytes, source_id: bytes, dst_id: bytes, run: dict[str, Any]
+        self, protocol: Any, system: str, pkt: bytes, source_id: bytes, dst_id: bytes, run: _PromptRun
     ) -> None:
         """Reactor side of ``play_on_slot``: one frame, unless the slot is taken."""
-        if run["stopped"]:
+        if run.stopped:
             return
         slot = protocol.STATUS.get(2) if getattr(protocol, "STATUS", None) else None
         if not slot:
-            run["stopped"] = True
+            run.stopped = True
             return
         stream_id = pkt[16:20]
         now = time.time()
-        if _slot_taken_by_other_stream(slot, stream_id, now):
-            run["stopped"] = True
-            logger.info(
-                "(%s) Voice on TS2, stopping server prompt after %s frames", system, run["sent"]
-            )
+        if slot_voice_held_by_other_stream(slot, stream_id, now):
+            run.stopped = True
+            logger.info("(%s) Voice on TS2, stopping server prompt after %s frames", system, run.sent)
             return
         slot["TX_TYPE"] = HBPF_SLT_VHEAD
         slot["TX_STREAM_ID"] = stream_id
         slot["TX_RFS"] = source_id
         slot["TX_TIME"] = now
-        run["stream_id"] = stream_id
+        run.stream_id = stream_id
         protocol.send_voice_packet(pkt, source_id, dst_id, slot)
-        run["sent"] += 1
+        run.sent += 1
 
-    def _prompt_end(self, protocol: Any, run: dict[str, Any]) -> None:
+    def _prompt_end(self, protocol: Any, run: _PromptRun) -> None:
         """Reactor side of ``play_on_slot``: free the slot if the prompt still holds it."""
         slot = protocol.STATUS.get(2) if getattr(protocol, "STATUS", None) else None
-        if slot and run["stream_id"] is not None and slot.get("TX_STREAM_ID") == run["stream_id"]:
+        if slot and run.stream_id is not None and slot.get("TX_STREAM_ID") == run.stream_id:
             slot["TX_TYPE"] = HBPF_SLT_VTERM
 
     def play_file_on_request(self, file_number: str, system: str) -> None:

tests/voice/test_prompt_holds_slot.py

@@ -1,5 +1,7 @@
 # ADN DMR Peer Server - tests server prompts hold the slot they play on
 #
+# Copyright (C) 2026  Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
+#
 ###############################################################################
 #   This program is free software; you can redistribute it and/or modify
 #   it under the terms of the GNU General Public License as published by