fix: route busy-steer voice through the shared out-of-band STT choke point

Follow-up for salvaged #65023/#53020: _prepare_busy_steer_text now calls
_transcribe_and_echo_pending_voice (the same helper the interrupt monitor
and pending-drain paths use) instead of a private transcription+echo copy,
so out-of-band voice pays one STT call per platform message and the echo
respects the count-based ledger from #67281. can_steer now accepts events
whose attachments are all STT-eligible voice media, completing the steer
half of #58780. Adds extract_media gating tests for #44826 and the
contributor mapping for chefboyrdave21.

contributors/emails/lumina@douno.it

@@ -0,0 +1 @@
+chefboyrdave21

gateway/run.py

@@ -6172,48 +6172,29 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         the automatic STT contract used by ``_prepare_inbound_message_text``.
         If transcription fails, preserve any caption and let the existing
         steer fallback handle an otherwise empty event without losing it.
+
+        Routes through ``_transcribe_and_echo_pending_voice`` — the single
+        out-of-band transcription choke point shared with the interrupt
+        monitor and the pending-drain path — so the STT call is made at most
+        once per platform message (cached on the event) and the transcript
+        echo respects the count-based ledger.  If steering later falls back
+        to queue mode, the drain path reuses the cached transcript instead of
+        paying for a second STT call or re-echoing the same line.
         """
         text = (event.text or "").strip()
-        media_urls = getattr(event, "media_urls", None) or []
-        media_types = getattr(event, "media_types", None) or []
-        voice_paths: List[str] = []
-
-        for index, path in enumerate(media_urls):
-            media_type = media_types[index] if index < len(media_types) else ""
-            is_voice = event.message_type == MessageType.VOICE or (
-                media_type.startswith("audio/")
-                and event.message_type not in {MessageType.AUDIO, MessageType.DOCUMENT}
-            )
-            if is_voice:
-                voice_paths.append(path)
-
-        if not voice_paths:
+        if not self._pending_event_audio_paths(event):
             return text
 
-        enriched_text, successful_transcripts = await self._enrich_message_with_transcription(
+        adapter = self._adapter_for_source(event.source)
+        enriched_text, successful_transcripts = await self._transcribe_and_echo_pending_voice(
+            event,
+            adapter,
+            event.source,
             text,
-            voice_paths,
+            log_context="Busy-steer",
         )
         if not successful_transcripts:
             return text
-
-        if self._should_echo_stt_transcripts():
-            adapter = self._adapter_for_source(event.source)
-            if adapter:
-                echo_meta = self._thread_metadata_for_source(
-                    event.source,
-                    self._reply_anchor_for_event(event),
-                )
-                for transcript in successful_transcripts:
-                    try:
-                        await adapter.send(
-                            event.source.chat_id,
-                            f'🎙️ "{transcript}"',
-                            metadata=echo_meta,
-                        )
-                    except Exception as exc:
-                        logger.debug("Busy-steer transcript echo failed (non-fatal): %s", exc)
-
         return (enriched_text or text).strip()
 
     async def _handle_active_session_busy_message(self, event: MessageEvent, session_key: str) -> bool:
@@ -6404,11 +6385,24 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         redirected = False
         if effective_mode == "steer":
             steer_text = await self._prepare_busy_steer_text(event)
+            # A follow-up qualifies for steering when it is plain text, OR
+            # when every attachment is STT-eligible voice media whose
+            # transcript was just folded into steer_text — otherwise a voice
+            # note in steer mode silently degrades to queue mode (#58780).
+            _steer_media_urls = getattr(event, "media_urls", None) or []
+            _steer_all_voice = bool(_steer_media_urls) and (
+                len(self._pending_event_audio_paths(event)) == len(_steer_media_urls)
+            )
             can_steer = (
                 steer_text
-                and event.message_type == MessageType.TEXT
-                and not event.media_urls
-                and not event.media_types
+                and (
+                    (
+                        event.message_type == MessageType.TEXT
+                        and not event.media_urls
+                        and not event.media_types
+                    )
+                    or _steer_all_voice
+                )
                 and running_agent is not None
                 and running_agent is not _AGENT_PENDING_SENTINEL
                 and hasattr(running_agent, "steer")

tests/gateway/test_platform_base.py

@@ -370,6 +370,29 @@ class TestExtractMedia:
         assert media[0][0] == "/path/to/voice.ogg"
         assert media[0][1] is True  # voice tag present
 
+    def test_voice_directive_only_taints_audio_files(self):
+        """[[audio_as_voice]] is message-global but must only flag audio files.
+
+        A non-audio file marked is_voice is excluded from the embedded-photo
+        batch and falls through to send_document, so an image sharing a
+        message with a voice note used to arrive as a file attachment
+        (#44826).
+        """
+        content = "[[audio_as_voice]]\nMEDIA:/tmp/pic.png\nMEDIA:/tmp/voice.ogg"
+        media, cleaned = BasePlatformAdapter.extract_media(content)
+        flags = dict(media)
+        assert flags["/tmp/pic.png"] is False
+        assert flags["/tmp/voice.ogg"] is True
+        assert "[[audio_as_voice]]" not in cleaned
+
+    def test_voice_directive_skips_video_and_documents(self):
+        content = "[[audio_as_voice]]\nMEDIA:/tmp/clip.mp4\nMEDIA:/tmp/report.pdf\nMEDIA:/tmp/note.opus"
+        media, _ = BasePlatformAdapter.extract_media(content)
+        flags = dict(media)
+        assert flags["/tmp/clip.mp4"] is False
+        assert flags["/tmp/report.pdf"] is False
+        assert flags["/tmp/note.opus"] is True
+
     def test_multiple_media_tags(self):
         content = "MEDIA:/a.ogg\nMEDIA:/b.ogg"
         media, _ = BasePlatformAdapter.extract_media(content)