Refactor messaging platform bridge (#858)

## Summary
- Extract shared messaging platform outbox for queued send/edit/delete,
dedup keys, limiter delegation, and fire-and-forget behavior
- Extract shared voice-note flow for pending voice state, transcription,
cancellation, cleanup, error replies, and IncomingMessage handoff
- Keep Telegram and Discord adapters focused on SDK lifecycle, event
extraction, attachment download, and raw send/edit/delete calls
- Document the split in ARCHITECTURE.md and bump version to 2.3.8

## Verification
- uv run pytest tests/messaging/test_platform_outbox.py
tests/messaging/test_platform_voice_flow.py
tests/messaging/test_voice_handlers.py tests/messaging/test_telegram.py
tests/messaging/test_telegram_edge_cases.py
tests/messaging/test_discord_platform.py
tests/contracts/test_import_boundaries.py
tests/contracts/test_architecture_contracts.py
- .\\scripts\\ci.ps1

ARCHITECTURE.md

@@ -488,6 +488,17 @@ Messaging is optional. [api/runtime.py](api/runtime.py) calls
 If `MESSAGING_PLATFORM` is `none`, or if the selected platform token is missing,
 the messaging bridge is skipped.
 
+Platform adapters in [messaging/platforms/telegram.py](messaging/platforms/telegram.py)
+and [messaging/platforms/discord.py](messaging/platforms/discord.py) are thin SDK
+shells: they own client lifecycle, event extraction, SDK retries, attachment
+download, and raw send/edit/delete calls. Shared delivery policy lives in
+[messaging/platforms/outbox.py](messaging/platforms/outbox.py), which owns queued
+send/edit/delete, dedup keys, limiter delegation, and fire-and-forget behavior.
+Shared voice-note orchestration lives in
+[messaging/platforms/voice_flow.py](messaging/platforms/voice_flow.py), which owns
+pending voice registration, temp-file cleanup, transcription, cancellation, error
+replies, and the handoff to `IncomingMessage`.
+
 [messaging/workflow.py](messaging/workflow.py) contains `MessagingWorkflow`, the
 platform-agnostic coordinator. It owns dependencies, callback wiring, stop/clear
 side effects, render settings, and shutdown-visible state.

messaging/platforms/discord.py

@@ -6,9 +6,7 @@ Implements MessagingPlatform for Discord using discord.py.
 
 import asyncio
 import contextlib
-import tempfile
 from collections.abc import Awaitable, Callable
-from pathlib import Path
 from typing import Any, cast
 
 from loguru import logger
@@ -17,10 +15,14 @@ from core.anthropic import format_user_error_preview
 
 from ..models import IncomingMessage
 from ..rendering.discord_markdown import format_status_discord
-from ..voice import PendingVoiceRegistry, VoiceTranscriptionService
 from .base import MessagingPlatform
-
-AUDIO_EXTENSIONS = (".ogg", ".mp4", ".mp3", ".wav", ".m4a")
+from .outbox import PlatformOutbox
+from .voice_flow import (
+    VoiceNoteFlow,
+    VoiceNoteRequest,
+    audio_suffix_from_metadata,
+    is_audio_metadata,
+)
 
 _discord_module: Any = None
 try:
@@ -124,14 +126,55 @@ class DiscordPlatform(MessagingPlatform):
         self._connected = False
         self._limiter: Any | None = None
         self._start_task: asyncio.Task | None = None
-        self._pending_voice = PendingVoiceRegistry()
-        self._voice_transcription = VoiceTranscriptionService(
+
+        async def send_operation(
+            chat_id: str,
+            text: str,
+            reply_to: str | None,
+            parse_mode: str | None,
+            message_thread_id: str | None,
+        ) -> str:
+            return await self.send_message(
+                chat_id,
+                text,
+                reply_to,
+                parse_mode,
+                message_thread_id,
+            )
+
+        async def edit_operation(
+            chat_id: str,
+            message_id: str,
+            text: str,
+            parse_mode: str | None,
+        ) -> None:
+            await self.edit_message(chat_id, message_id, text, parse_mode)
+
+        async def delete_operation(chat_id: str, message_id: str) -> None:
+            await self.delete_message(chat_id, message_id)
+
+        async def delete_many_operation(
+            chat_id: str,
+            message_ids: list[str],
+        ) -> None:
+            await self.delete_messages(chat_id, message_ids)
+
+        self._outbox = PlatformOutbox(
+            get_limiter=lambda: self._limiter,
+            send=send_operation,
+            edit=edit_operation,
+            delete=delete_operation,
+            delete_many=delete_many_operation,
+        )
+        self._voice_flow = VoiceNoteFlow(
+            voice_note_enabled=voice_note_enabled,
+            whisper_model=whisper_model,
+            whisper_device=whisper_device,
             hf_token=hf_token,
             nvidia_nim_api_key=nvidia_nim_api_key,
+            log_raw_messaging_content=log_raw_messaging_content,
+            log_api_error_tracebacks=log_api_error_tracebacks,
         )
-        self._voice_note_enabled = voice_note_enabled
-        self._whisper_model = whisper_model
-        self._whisper_device = whisper_device
         self._messaging_rate_limit = messaging_rate_limit
         self._messaging_rate_window = messaging_rate_window
         self._log_raw_messaging_content = log_raw_messaging_content
@@ -145,26 +188,26 @@ class DiscordPlatform(MessagingPlatform):
         self, chat_id: str, voice_msg_id: str, status_msg_id: str
     ) -> None:
         """Register a voice note as pending transcription."""
-        await self._pending_voice.register(chat_id, voice_msg_id, status_msg_id)
+        await self._voice_flow.register_pending_voice(
+            chat_id,
+            voice_msg_id,
+            status_msg_id,
+        )
 
     async def cancel_pending_voice(
         self, chat_id: str, reply_id: str
     ) -> tuple[str, str] | None:
         """Cancel a pending voice transcription. Returns (voice_msg_id, status_msg_id) if found."""
-        return await self._pending_voice.cancel(chat_id, reply_id)
+        return await self._voice_flow.cancel_pending_voice(chat_id, reply_id)
 
     async def _is_voice_still_pending(self, chat_id: str, voice_msg_id: str) -> bool:
         """Check if a voice note is still pending (not cancelled)."""
-        return await self._pending_voice.is_pending(chat_id, voice_msg_id)
+        return await self._voice_flow.is_voice_still_pending(chat_id, voice_msg_id)
 
     def _get_audio_attachment(self, message: Any) -> Any | None:
         """Return first audio attachment, or None."""
         for att in message.attachments:
-            ct = (att.content_type or "").lower()
-            fn = (att.filename or "").lower()
-            if ct.startswith("audio/") or any(
-                fn.endswith(ext) for ext in AUDIO_EXTENSIONS
-            ):
+            if is_audio_metadata(att.filename, att.content_type):
                 return att
         return None
 
@@ -172,115 +215,43 @@ class DiscordPlatform(MessagingPlatform):
         self, message: Any, attachment: Any, channel_id: str
     ) -> bool:
         """Handle voice/audio attachment. Returns True if handled."""
-        if not self._voice_note_enabled:
-            await message.reply("Voice notes are disabled.")
-            return True
-
-        if not self._message_handler:
-            return False
-
-        status_msg_id = await self.queue_send_message(
-            channel_id,
-            format_status_discord("Transcribing voice note..."),
-            reply_to=str(message.id),
-            fire_and_forget=False,
-        )
-
         user_id = str(message.author.id)
         message_id = str(message.id)
-        await self._register_pending_voice(channel_id, message_id, str(status_msg_id))
         reply_to = (
             str(message.reference.message_id)
             if message.reference and message.reference.message_id
             else None
         )
-
-        ext = ".ogg"
-        fn = (attachment.filename or "").lower()
-        for e in AUDIO_EXTENSIONS:
-            if fn.endswith(e):
-                ext = e
-                break
         ct = attachment.content_type or "audio/ogg"
-        if "mp4" in ct or "m4a" in fn:
-            ext = ".m4a" if "m4a" in fn else ".mp4"
-        elif "mp3" in ct or fn.endswith(".mp3"):
-            ext = ".mp3"
 
-        with tempfile.NamedTemporaryFile(suffix=ext, delete=False) as tmp:
-            tmp_path = Path(tmp.name)
-
-        try:
+        async def _download_to(tmp_path) -> None:
             await attachment.save(str(tmp_path))
 
-            transcribed = await self._voice_transcription.transcribe(
-                tmp_path,
-                ct,
-                whisper_model=self._whisper_model,
-                whisper_device=self._whisper_device,
-            )
+        async def _reply_text(text: str) -> None:
+            await message.reply(text)
 
-            if not await self._is_voice_still_pending(channel_id, message_id):
-                await self.queue_delete_message(channel_id, str(status_msg_id))
-                return True
-
-            await self._pending_voice.complete(
-                channel_id, message_id, str(status_msg_id)
-            )
-
-            incoming = IncomingMessage(
-                text=transcribed,
+        return await self._voice_flow.handle(
+            VoiceNoteRequest(
+                platform="discord",
                 chat_id=channel_id,
                 user_id=user_id,
                 message_id=message_id,
-                platform="discord",
+                raw_event=message,
+                content_type=ct,
+                temp_suffix=audio_suffix_from_metadata(
+                    filename=attachment.filename,
+                    content_type=attachment.content_type,
+                ),
+                status_text=format_status_discord("Transcribing voice note..."),
                 reply_to_message_id=reply_to,
                 username=message.author.display_name,
-                raw_event=message,
-                status_message_id=status_msg_id,
-            )
-
-            if self._log_raw_messaging_content:
-                logger.info(
-                    "DISCORD_VOICE: chat_id={} message_id={} transcribed={!r}",
-                    channel_id,
-                    message_id,
-                    (
-                        transcribed[:80] + "..."
-                        if len(transcribed) > 80
-                        else transcribed
-                    ),
-                )
-            else:
-                logger.info(
-                    "DISCORD_VOICE: chat_id={} message_id={} transcribed_len={}",
-                    channel_id,
-                    message_id,
-                    len(transcribed),
-                )
-
-            await self._message_handler(incoming)
-            return True
-        except ValueError as e:
-            await message.reply(format_user_error_preview(e))
-            return True
-        except ImportError as e:
-            await message.reply(format_user_error_preview(e))
-            return True
-        except Exception as e:
-            if self._log_api_error_tracebacks:
-                logger.error("Voice transcription failed: {}", e)
-            else:
-                logger.error(
-                    "Voice transcription failed: exc_type={}", type(e).__name__
-                )
-            await message.reply(
-                "Could not transcribe voice note. Please try again or send text."
-            )
-            return True
-        finally:
-            with contextlib.suppress(OSError):
-                tmp_path.unlink(missing_ok=True)
+                download_to=_download_to,
+                reply_text=_reply_text,
+            ),
+            message_handler=self._message_handler,
+            queue_send_message=self.queue_send_message,
+            queue_delete_message=self.queue_delete_message,
+        )
 
     async def _on_discord_message(self, message: Any) -> None:
         """Handle incoming Discord messages."""
@@ -409,7 +380,7 @@ class DiscordPlatform(MessagingPlatform):
         self._connected = False
         logger.info("Discord platform stopped")
 
-    async def send_message(
+    async def _send_message_raw(
         self,
         chat_id: str,
         text: str,
@@ -437,7 +408,24 @@ class DiscordPlatform(MessagingPlatform):
 
         return str(msg.id)
 
-    async def edit_message(
+    async def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        reply_to: str | None = None,
+        parse_mode: str | None = None,
+        message_thread_id: str | None = None,
+    ) -> str:
+        """Send a message to a channel."""
+        return await self._send_message_raw(
+            chat_id,
+            text,
+            reply_to,
+            parse_mode,
+            message_thread_id,
+        )
+
+    async def _edit_message_raw(
         self,
         chat_id: str,
         message_id: str,
@@ -459,7 +447,17 @@ class DiscordPlatform(MessagingPlatform):
         text = self._truncate(text)
         await msg.edit(content=text)
 
-    async def delete_message(
+    async def edit_message(
+        self,
+        chat_id: str,
+        message_id: str,
+        text: str,
+        parse_mode: str | None = None,
+    ) -> None:
+        """Edit an existing message."""
+        await self._edit_message_raw(chat_id, message_id, text, parse_mode)
+
+    async def _delete_message_raw(
         self,
         chat_id: str,
         message_id: str,
@@ -477,10 +475,22 @@ class DiscordPlatform(MessagingPlatform):
         except discord.NotFound, discord.Forbidden:
             pass
 
-    async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
+    async def delete_message(
+        self,
+        chat_id: str,
+        message_id: str,
+    ) -> None:
+        """Delete a message from a channel."""
+        await self._delete_message_raw(chat_id, message_id)
+
+    async def _delete_messages_raw(self, chat_id: str, message_ids: list[str]) -> None:
         """Delete multiple messages (best-effort)."""
         for mid in message_ids:
-            await self.delete_message(chat_id, mid)
+            await self._delete_message_raw(chat_id, mid)
+
+    async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
+        """Delete multiple messages (best-effort)."""
+        await self._delete_messages_raw(chat_id, message_ids)
 
     async def queue_send_message(
         self,
@@ -492,20 +502,14 @@ class DiscordPlatform(MessagingPlatform):
         message_thread_id: str | None = None,
     ) -> str | None:
         """Enqueue a message to be sent."""
-        if not self._limiter:
-            return await self.send_message(
-                chat_id, text, reply_to, parse_mode, message_thread_id
-            )
-
-        async def _send():
-            return await self.send_message(
-                chat_id, text, reply_to, parse_mode, message_thread_id
-            )
-
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_send)
-            return None
-        return await self._limiter.enqueue(_send)
+        return await self._outbox.queue_send_message(
+            chat_id,
+            text,
+            reply_to,
+            parse_mode,
+            fire_and_forget,
+            message_thread_id,
+        )
 
     async def queue_edit_message(
         self,
@@ -516,18 +520,13 @@ class DiscordPlatform(MessagingPlatform):
         fire_and_forget: bool = True,
     ) -> None:
         """Enqueue a message edit."""
-        if not self._limiter:
-            await self.edit_message(chat_id, message_id, text, parse_mode)
-            return
-
-        async def _edit():
-            await self.edit_message(chat_id, message_id, text, parse_mode)
-
-        dedup_key = f"edit:{chat_id}:{message_id}"
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_edit, dedup_key=dedup_key)
-        else:
-            await self._limiter.enqueue(_edit, dedup_key=dedup_key)
+        await self._outbox.queue_edit_message(
+            chat_id,
+            message_id,
+            text,
+            parse_mode,
+            fire_and_forget,
+        )
 
     async def queue_delete_message(
         self,
@@ -536,18 +535,7 @@ class DiscordPlatform(MessagingPlatform):
         fire_and_forget: bool = True,
     ) -> None:
         """Enqueue a message delete."""
-        if not self._limiter:
-            await self.delete_message(chat_id, message_id)
-            return
-
-        async def _delete():
-            await self.delete_message(chat_id, message_id)
-
-        dedup_key = f"del:{chat_id}:{message_id}"
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_delete, dedup_key=dedup_key)
-        else:
-            await self._limiter.enqueue(_delete, dedup_key=dedup_key)
+        await self._outbox.queue_delete_message(chat_id, message_id, fire_and_forget)
 
     async def queue_delete_messages(
         self,
@@ -556,28 +544,15 @@ class DiscordPlatform(MessagingPlatform):
         fire_and_forget: bool = True,
     ) -> None:
         """Enqueue a bulk delete."""
-        if not message_ids:
-            return
-
-        if not self._limiter:
-            await self.delete_messages(chat_id, message_ids)
-            return
-
-        async def _bulk():
-            await self.delete_messages(chat_id, message_ids)
-
-        dedup_key = f"del_bulk:{chat_id}:{hash(tuple(message_ids))}"
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_bulk, dedup_key=dedup_key)
-        else:
-            await self._limiter.enqueue(_bulk, dedup_key=dedup_key)
+        await self._outbox.queue_delete_messages(
+            chat_id,
+            message_ids,
+            fire_and_forget,
+        )
 
     def fire_and_forget(self, task: Awaitable[Any]) -> None:
         """Execute a coroutine without awaiting it."""
-        if asyncio.iscoroutine(task):
-            _ = asyncio.create_task(task)
-        else:
-            _ = asyncio.ensure_future(task)
+        self._outbox.fire_and_forget(task)
 
     def on_message(
         self,

messaging/platforms/outbox.py

@@ -0,0 +1,144 @@
+"""Shared queued delivery helper for messaging platforms."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Any, cast
+
+SendOperation = Callable[
+    [str, str, str | None, str | None, str | None],
+    Awaitable[str],
+]
+EditOperation = Callable[[str, str, str, str | None], Awaitable[None]]
+DeleteOperation = Callable[[str, str], Awaitable[None]]
+DeleteManyOperation = Callable[[str, list[str]], Awaitable[None]]
+LimiterGetter = Callable[[], Any | None]
+
+
+class PlatformOutbox:
+    """Own queueing, deduplication, and fire-and-forget delivery policy."""
+
+    def __init__(
+        self,
+        *,
+        get_limiter: LimiterGetter,
+        send: SendOperation,
+        edit: EditOperation,
+        delete: DeleteOperation,
+        delete_many: DeleteManyOperation,
+    ) -> None:
+        self._get_limiter = get_limiter
+        self._send = send
+        self._edit = edit
+        self._delete = delete
+        self._delete_many = delete_many
+
+    async def queue_send_message(
+        self,
+        chat_id: str,
+        text: str,
+        reply_to: str | None = None,
+        parse_mode: str | None = None,
+        fire_and_forget: bool = True,
+        message_thread_id: str | None = None,
+    ) -> str | None:
+        """Queue or immediately send a platform message."""
+        limiter = self._get_limiter()
+        if limiter is None:
+            return await self._send(
+                chat_id,
+                text,
+                reply_to,
+                parse_mode,
+                message_thread_id,
+            )
+
+        async def _send() -> str:
+            return await self._send(
+                chat_id,
+                text,
+                reply_to,
+                parse_mode,
+                message_thread_id,
+            )
+
+        if fire_and_forget:
+            limiter.fire_and_forget(_send)
+            return None
+        return cast(str | None, await limiter.enqueue(_send))
+
+    async def queue_edit_message(
+        self,
+        chat_id: str,
+        message_id: str,
+        text: str,
+        parse_mode: str | None = None,
+        fire_and_forget: bool = True,
+    ) -> None:
+        """Queue or immediately edit a platform message."""
+        limiter = self._get_limiter()
+        if limiter is None:
+            await self._edit(chat_id, message_id, text, parse_mode)
+            return
+
+        async def _edit() -> None:
+            await self._edit(chat_id, message_id, text, parse_mode)
+
+        dedup_key = f"edit:{chat_id}:{message_id}"
+        if fire_and_forget:
+            limiter.fire_and_forget(_edit, dedup_key=dedup_key)
+        else:
+            await limiter.enqueue(_edit, dedup_key=dedup_key)
+
+    async def queue_delete_message(
+        self,
+        chat_id: str,
+        message_id: str,
+        fire_and_forget: bool = True,
+    ) -> None:
+        """Queue or immediately delete a platform message."""
+        limiter = self._get_limiter()
+        if limiter is None:
+            await self._delete(chat_id, message_id)
+            return
+
+        async def _delete() -> None:
+            await self._delete(chat_id, message_id)
+
+        dedup_key = f"del:{chat_id}:{message_id}"
+        if fire_and_forget:
+            limiter.fire_and_forget(_delete, dedup_key=dedup_key)
+        else:
+            await limiter.enqueue(_delete, dedup_key=dedup_key)
+
+    async def queue_delete_messages(
+        self,
+        chat_id: str,
+        message_ids: list[str],
+        fire_and_forget: bool = True,
+    ) -> None:
+        """Queue or immediately bulk-delete platform messages."""
+        if not message_ids:
+            return
+
+        limiter = self._get_limiter()
+        if limiter is None:
+            await self._delete_many(chat_id, message_ids)
+            return
+
+        async def _delete_many() -> None:
+            await self._delete_many(chat_id, message_ids)
+
+        dedup_key = f"del_bulk:{chat_id}:{hash(tuple(message_ids))}"
+        if fire_and_forget:
+            limiter.fire_and_forget(_delete_many, dedup_key=dedup_key)
+        else:
+            await limiter.enqueue(_delete_many, dedup_key=dedup_key)
+
+    def fire_and_forget(self, task: Awaitable[Any]) -> None:
+        """Execute a coroutine or future without awaiting it."""
+        if asyncio.iscoroutine(task):
+            _ = asyncio.create_task(task)
+        else:
+            _ = asyncio.ensure_future(task)

messaging/platforms/telegram.py

@@ -7,8 +7,6 @@ Implements MessagingPlatform for Telegram using python-telegram-bot.
 import asyncio
 import contextlib
 import os
-import tempfile
-from pathlib import Path
 
 # Opt-in to future behavior for python-telegram-bot (retry_after as timedelta)
 # This must be set BEFORE importing telegram.error
@@ -27,8 +25,9 @@ if TYPE_CHECKING:
 
 from ..models import IncomingMessage
 from ..rendering.telegram_markdown import escape_md_v2, format_status
-from ..voice import PendingVoiceRegistry, VoiceTranscriptionService
 from .base import MessagingPlatform
+from .outbox import PlatformOutbox
+from .voice_flow import VoiceNoteFlow, VoiceNoteRequest, audio_suffix_from_metadata
 
 # Optional import - python-telegram-bot may not be installed
 try:
@@ -92,15 +91,55 @@ class TelegramPlatform(MessagingPlatform):
         )
         self._connected = False
         self._limiter: Any | None = None  # Will be MessagingRateLimiter
-        # Pending voice transcriptions: (chat_id, msg_id) -> (voice_msg_id, status_msg_id)
-        self._pending_voice = PendingVoiceRegistry()
-        self._voice_transcription = VoiceTranscriptionService(
+
+        async def send_operation(
+            chat_id: str,
+            text: str,
+            reply_to: str | None,
+            parse_mode: str | None,
+            message_thread_id: str | None,
+        ) -> str:
+            return await self.send_message(
+                chat_id,
+                text,
+                reply_to,
+                parse_mode,
+                message_thread_id,
+            )
+
+        async def edit_operation(
+            chat_id: str,
+            message_id: str,
+            text: str,
+            parse_mode: str | None,
+        ) -> None:
+            await self.edit_message(chat_id, message_id, text, parse_mode)
+
+        async def delete_operation(chat_id: str, message_id: str) -> None:
+            await self.delete_message(chat_id, message_id)
+
+        async def delete_many_operation(
+            chat_id: str,
+            message_ids: list[str],
+        ) -> None:
+            await self.delete_messages(chat_id, message_ids)
+
+        self._outbox = PlatformOutbox(
+            get_limiter=lambda: self._limiter,
+            send=send_operation,
+            edit=edit_operation,
+            delete=delete_operation,
+            delete_many=delete_many_operation,
+        )
+        self._voice_flow = VoiceNoteFlow(
+            voice_note_enabled=voice_note_enabled,
+            whisper_model=whisper_model,
+            whisper_device=whisper_device,
             hf_token=hf_token,
             nvidia_nim_api_key=nvidia_nim_api_key,
+            log_raw_messaging_content=log_raw_messaging_content,
+            log_api_error_tracebacks=log_api_error_tracebacks,
         )
-        self._voice_note_enabled = voice_note_enabled
-        self._whisper_model = whisper_model
-        self._whisper_device = whisper_device
         self._messaging_rate_limit = messaging_rate_limit
         self._messaging_rate_window = messaging_rate_window
         self._log_raw_messaging_content = log_raw_messaging_content
@@ -110,17 +149,21 @@ class TelegramPlatform(MessagingPlatform):
         self, chat_id: str, voice_msg_id: str, status_msg_id: str
     ) -> None:
         """Register a voice note as pending transcription (for /clear reply during transcription)."""
-        await self._pending_voice.register(chat_id, voice_msg_id, status_msg_id)
+        await self._voice_flow.register_pending_voice(
+            chat_id,
+            voice_msg_id,
+            status_msg_id,
+        )
 
     async def cancel_pending_voice(
         self, chat_id: str, reply_id: str
     ) -> tuple[str, str] | None:
         """Cancel a pending voice transcription. Returns (voice_msg_id, status_msg_id) if found."""
-        return await self._pending_voice.cancel(chat_id, reply_id)
+        return await self._voice_flow.cancel_pending_voice(chat_id, reply_id)
 
     async def _is_voice_still_pending(self, chat_id: str, voice_msg_id: str) -> bool:
         """Check if a voice note is still pending (not cancelled)."""
-        return await self._pending_voice.is_pending(chat_id, voice_msg_id)
+        return await self._voice_flow.is_voice_still_pending(chat_id, voice_msg_id)
 
     async def start(self) -> None:
         """Initialize and connect to Telegram."""
@@ -277,7 +320,7 @@ class TelegramPlatform(MessagingPlatform):
                     return await func(*args, **kwargs)
                 raise
 
-    async def send_message(
+    async def _send_message_raw(
         self,
         chat_id: str,
         text: str,
@@ -305,7 +348,24 @@ class TelegramPlatform(MessagingPlatform):
 
         return await self._with_retry(_do_send, parse_mode=parse_mode)
 
-    async def edit_message(
+    async def send_message(
+        self,
+        chat_id: str,
+        text: str,
+        reply_to: str | None = None,
+        parse_mode: str | None = "MarkdownV2",
+        message_thread_id: str | None = None,
+    ) -> str:
+        """Send a message to a chat."""
+        return await self._send_message_raw(
+            chat_id,
+            text,
+            reply_to,
+            parse_mode,
+            message_thread_id,
+        )
+
+    async def _edit_message_raw(
         self,
         chat_id: str,
         message_id: str,
@@ -328,7 +388,17 @@ class TelegramPlatform(MessagingPlatform):
 
         await self._with_retry(_do_edit, parse_mode=parse_mode)
 
-    async def delete_message(
+    async def edit_message(
+        self,
+        chat_id: str,
+        message_id: str,
+        text: str,
+        parse_mode: str | None = "MarkdownV2",
+    ) -> None:
+        """Edit an existing message."""
+        await self._edit_message_raw(chat_id, message_id, text, parse_mode)
+
+    async def _delete_message_raw(
         self,
         chat_id: str,
         message_id: str,
@@ -344,7 +414,15 @@ class TelegramPlatform(MessagingPlatform):
 
         await self._with_retry(_do_delete)
 
-    async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
+    async def delete_message(
+        self,
+        chat_id: str,
+        message_id: str,
+    ) -> None:
+        """Delete a message from a chat."""
+        await self._delete_message_raw(chat_id, message_id)
+
+    async def _delete_messages_raw(self, chat_id: str, message_ids: list[str]) -> None:
         """Delete multiple messages (best-effort)."""
         if not message_ids:
             return
@@ -372,7 +450,11 @@ class TelegramPlatform(MessagingPlatform):
             return
 
         for mid in message_ids:
-            await self.delete_message(chat_id, mid)
+            await self._delete_message_raw(chat_id, mid)
+
+    async def delete_messages(self, chat_id: str, message_ids: list[str]) -> None:
+        """Delete multiple messages (best-effort)."""
+        await self._delete_messages_raw(chat_id, message_ids)
 
     async def queue_send_message(
         self,
@@ -384,22 +466,14 @@ class TelegramPlatform(MessagingPlatform):
         message_thread_id: str | None = None,
     ) -> str | None:
         """Enqueue a message to be sent (using limiter)."""
-        # Note: Bot API handles limits better, but we still use our limiter for nice queuing
-        if not self._limiter:
-            return await self.send_message(
-                chat_id, text, reply_to, parse_mode, message_thread_id
-            )
-
-        async def _send():
-            return await self.send_message(
-                chat_id, text, reply_to, parse_mode, message_thread_id
-            )
-
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_send)
-            return None
-        else:
-            return await self._limiter.enqueue(_send)
+        return await self._outbox.queue_send_message(
+            chat_id,
+            text,
+            reply_to,
+            parse_mode,
+            fire_and_forget,
+            message_thread_id,
+        )
 
     async def queue_edit_message(
         self,
@@ -410,17 +484,13 @@ class TelegramPlatform(MessagingPlatform):
         fire_and_forget: bool = True,
     ) -> None:
         """Enqueue a message edit."""
-        if not self._limiter:
-            return await self.edit_message(chat_id, message_id, text, parse_mode)
-
-        async def _edit():
-            return await self.edit_message(chat_id, message_id, text, parse_mode)
-
-        dedup_key = f"edit:{chat_id}:{message_id}"
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_edit, dedup_key=dedup_key)
-        else:
-            await self._limiter.enqueue(_edit, dedup_key=dedup_key)
+        await self._outbox.queue_edit_message(
+            chat_id,
+            message_id,
+            text,
+            parse_mode,
+            fire_and_forget,
+        )
 
     async def queue_delete_message(
         self,
@@ -429,17 +499,7 @@ class TelegramPlatform(MessagingPlatform):
         fire_and_forget: bool = True,
     ) -> None:
         """Enqueue a message delete."""
-        if not self._limiter:
-            return await self.delete_message(chat_id, message_id)
-
-        async def _delete():
-            return await self.delete_message(chat_id, message_id)
-
-        dedup_key = f"del:{chat_id}:{message_id}"
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_delete, dedup_key=dedup_key)
-        else:
-            await self._limiter.enqueue(_delete, dedup_key=dedup_key)
+        await self._outbox.queue_delete_message(chat_id, message_id, fire_and_forget)
 
     async def queue_delete_messages(
         self,
@@ -448,28 +508,15 @@ class TelegramPlatform(MessagingPlatform):
         fire_and_forget: bool = True,
     ) -> None:
         """Enqueue a bulk delete (if supported) or a sequence of deletes."""
-        if not message_ids:
-            return
-
-        if not self._limiter:
-            return await self.delete_messages(chat_id, message_ids)
-
-        async def _bulk():
-            return await self.delete_messages(chat_id, message_ids)
-
-        # Dedup by the chunk content; okay to be coarse here.
-        dedup_key = f"del_bulk:{chat_id}:{hash(tuple(message_ids))}"
-        if fire_and_forget:
-            self._limiter.fire_and_forget(_bulk, dedup_key=dedup_key)
-        else:
-            await self._limiter.enqueue(_bulk, dedup_key=dedup_key)
+        await self._outbox.queue_delete_messages(
+            chat_id,
+            message_ids,
+            fire_and_forget,
+        )
 
     def fire_and_forget(self, task: Awaitable[Any]) -> None:
         """Execute a coroutine without awaiting it."""
-        if asyncio.iscoroutine(task):
-            _ = asyncio.create_task(task)
-        else:
-            _ = asyncio.ensure_future(task)
+        self._outbox.fire_and_forget(task)
 
     def on_message(
         self,
@@ -578,123 +625,64 @@ class TelegramPlatform(MessagingPlatform):
         self, update: Update, context: ContextTypes.DEFAULT_TYPE
     ) -> None:
         """Handle incoming voice messages."""
+        message = update.message
+        effective_user = update.effective_user
+        effective_chat = update.effective_chat
         if (
-            not update.message
-            or not update.message.voice
-            or not update.effective_user
-            or not update.effective_chat
+            message is None
+            or message.voice is None
+            or effective_user is None
+            or effective_chat is None
         ):
             return
+        voice = message.voice
 
-        if not self._voice_note_enabled:
-            await update.message.reply_text("Voice notes are disabled.")
+        async def _reply_text(text: str) -> None:
+            await message.reply_text(text)
+
+        if await self._voice_flow.reply_if_disabled(_reply_text):
             return
 
-        user_id = str(update.effective_user.id)
-        chat_id = str(update.effective_chat.id)
+        user_id = str(effective_user.id)
+        chat_id = str(effective_chat.id)
 
         if self.allowed_user_id and user_id != str(self.allowed_user_id).strip():
             logger.warning(f"Unauthorized voice access attempt from {user_id}")
             return
 
-        if not self._message_handler:
-            return
-
         thread_id = (
-            str(update.message.message_thread_id)
-            if getattr(update.message, "message_thread_id", None) is not None
+            str(message.message_thread_id)
+            if getattr(message, "message_thread_id", None) is not None
             else None
         )
-        status_msg_id = await self.queue_send_message(
-            chat_id,
-            format_status("⏳", "Transcribing voice note..."),
-            reply_to=str(update.message.message_id),
-            parse_mode="MarkdownV2",
-            fire_and_forget=False,
-            message_thread_id=thread_id,
-        )
-
-        message_id = str(update.message.message_id)
-        await self._register_pending_voice(chat_id, message_id, str(status_msg_id))
+        message_id = str(message.message_id)
         reply_to = (
-            str(update.message.reply_to_message.message_id)
-            if update.message.reply_to_message
+            str(message.reply_to_message.message_id)
+            if message.reply_to_message
             else None
         )
 
-        voice = update.message.voice
-        suffix = ".ogg"
-        if voice.mime_type and "mpeg" in voice.mime_type:
-            suffix = ".mp3"
-        elif voice.mime_type and "mp4" in voice.mime_type:
-            suffix = ".mp4"
-
-        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
-            tmp_path = Path(tmp.name)
-
-        try:
+        async def _download_to(tmp_path) -> None:
             tg_file = await context.bot.get_file(voice.file_id)
             await tg_file.download_to_drive(custom_path=str(tmp_path))
 
-            transcribed = await self._voice_transcription.transcribe(
-                tmp_path,
-                voice.mime_type or "audio/ogg",
-                whisper_model=self._whisper_model,
-                whisper_device=self._whisper_device,
-            )
-
-            if not await self._is_voice_still_pending(chat_id, message_id):
-                await self.queue_delete_message(chat_id, str(status_msg_id))
-                return
-
-            await self._pending_voice.complete(chat_id, message_id, str(status_msg_id))
-
-            incoming = IncomingMessage(
-                text=transcribed,
+        await self._voice_flow.handle(
+            VoiceNoteRequest(
+                platform="telegram",
                 chat_id=chat_id,
                 user_id=user_id,
                 message_id=message_id,
-                platform="telegram",
-                reply_to_message_id=reply_to,
-                message_thread_id=thread_id,
                 raw_event=update,
-                status_message_id=status_msg_id,
-            )
-
-            if self._log_raw_messaging_content:
-                logger.info(
-                    "TELEGRAM_VOICE: chat_id={} message_id={} transcribed={!r}",
-                    chat_id,
-                    message_id,
-                    (
-                        transcribed[:80] + "..."
-                        if len(transcribed) > 80
-                        else transcribed
-                    ),
-                )
-            else:
-                logger.info(
-                    "TELEGRAM_VOICE: chat_id={} message_id={} transcribed_len={}",
-                    chat_id,
-                    message_id,
-                    len(transcribed),
-                )
-
-            await self._message_handler(incoming)
-        except ValueError as e:
-            await update.message.reply_text(format_user_error_preview(e))
-        except ImportError as e:
-            await update.message.reply_text(format_user_error_preview(e))
-        except Exception as e:
-            if self._log_api_error_tracebacks:
-                logger.error("Voice transcription failed: {}", e)
-            else:
-                logger.error(
-                    "Voice transcription failed: exc_type={}", type(e).__name__
-                )
-            await update.message.reply_text(
-                "Could not transcribe voice note. Please try again or send text."
-            )
-        finally:
-            with contextlib.suppress(OSError):
-                tmp_path.unlink(missing_ok=True)
+                content_type=voice.mime_type or "audio/ogg",
+                temp_suffix=audio_suffix_from_metadata(content_type=voice.mime_type),
+                status_text=format_status("⏳", "Transcribing voice note..."),
+                status_parse_mode="MarkdownV2",
+                message_thread_id=thread_id,
+                reply_to_message_id=reply_to,
+                download_to=_download_to,
+                reply_text=_reply_text,
+            ),
+            message_handler=self._message_handler,
+            queue_send_message=self.queue_send_message,
+            queue_delete_message=self.queue_delete_message,
+        )

messaging/platforms/voice_flow.py

@@ -0,0 +1,295 @@
+"""Shared voice-note flow for messaging platform adapters."""
+
+from __future__ import annotations
+
+import contextlib
+import tempfile
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from loguru import logger
+
+from core.anthropic import format_user_error_preview
+
+from ..models import IncomingMessage
+from ..voice import PendingVoiceRegistry, VoiceTranscriptionService
+
+AUDIO_EXTENSIONS = (".ogg", ".mp4", ".mp3", ".wav", ".m4a")
+VOICE_DISABLED_MESSAGE = "Voice notes are disabled."
+VOICE_TRANSCRIPTION_ERROR_MESSAGE = (
+    "Could not transcribe voice note. Please try again or send text."
+)
+
+MessageHandler = Callable[[IncomingMessage], Awaitable[None]]
+QueueSend = Callable[..., Awaitable[str | None]]
+QueueDelete = Callable[..., Awaitable[None]]
+
+
+@dataclass(frozen=True)
+class VoiceNoteRequest:
+    """Platform-normalized voice-note input."""
+
+    platform: str
+    chat_id: str
+    user_id: str
+    message_id: str
+    raw_event: Any
+    content_type: str
+    temp_suffix: str
+    status_text: str
+    download_to: Callable[[Path], Awaitable[None]]
+    reply_text: Callable[[str], Awaitable[None]]
+    reply_to_message_id: str | None = None
+    status_parse_mode: str | None = None
+    message_thread_id: str | None = None
+    username: str | None = None
+
+
+def is_audio_metadata(filename: str | None, content_type: str | None) -> bool:
+    """Return whether attachment metadata describes an audio file."""
+    normalized_content_type = (content_type or "").lower()
+    normalized_filename = (filename or "").lower()
+    return normalized_content_type.startswith("audio/") or any(
+        normalized_filename.endswith(extension) for extension in AUDIO_EXTENSIONS
+    )
+
+
+def audio_suffix_from_metadata(
+    *,
+    filename: str | None = None,
+    content_type: str | None = None,
+    default: str = ".ogg",
+) -> str:
+    """Choose a temp-file suffix from platform attachment metadata."""
+    normalized_filename = (filename or "").lower()
+    normalized_content_type = (content_type or "").lower()
+
+    if "m4a" in normalized_content_type:
+        return ".m4a"
+    if "mp4" in normalized_content_type:
+        if normalized_filename.endswith(".m4a"):
+            return ".m4a"
+        return ".mp4"
+    if "mpeg" in normalized_content_type or "mp3" in normalized_content_type:
+        return ".mp3"
+    if "wav" in normalized_content_type:
+        return ".wav"
+
+    for extension in AUDIO_EXTENSIONS:
+        if normalized_filename.endswith(extension):
+            return extension
+    return default
+
+
+class VoiceNoteFlow:
+    """Own common voice transcription state and control flow."""
+
+    def __init__(
+        self,
+        *,
+        voice_note_enabled: bool,
+        whisper_model: str,
+        whisper_device: str,
+        hf_token: str,
+        nvidia_nim_api_key: str,
+        log_raw_messaging_content: bool,
+        log_api_error_tracebacks: bool,
+    ) -> None:
+        self._voice_note_enabled = voice_note_enabled
+        self._whisper_model = whisper_model
+        self._whisper_device = whisper_device
+        self._log_raw_messaging_content = log_raw_messaging_content
+        self._log_api_error_tracebacks = log_api_error_tracebacks
+        self._pending_voice = PendingVoiceRegistry()
+        self._voice_transcription = VoiceTranscriptionService(
+            hf_token=hf_token,
+            nvidia_nim_api_key=nvidia_nim_api_key,
+        )
+
+    @property
+    def is_enabled(self) -> bool:
+        """Return whether voice-note handling is enabled."""
+        return self._voice_note_enabled
+
+    async def reply_if_disabled(
+        self, reply_text: Callable[[str], Awaitable[None]]
+    ) -> bool:
+        """Reply with the disabled message when voice-note handling is disabled."""
+        if self._voice_note_enabled:
+            return False
+        await reply_text(VOICE_DISABLED_MESSAGE)
+        return True
+
+    async def register_pending_voice(
+        self, chat_id: str, voice_msg_id: str, status_msg_id: str
+    ) -> None:
+        """Register a voice note as pending transcription."""
+        await self._pending_voice.register(chat_id, voice_msg_id, status_msg_id)
+
+    async def cancel_pending_voice(
+        self, chat_id: str, reply_id: str
+    ) -> tuple[str, str] | None:
+        """Cancel a pending voice transcription."""
+        return await self._pending_voice.cancel(chat_id, reply_id)
+
+    async def is_voice_still_pending(self, chat_id: str, voice_msg_id: str) -> bool:
+        """Return whether a voice note is still pending."""
+        return await self._pending_voice.is_pending(chat_id, voice_msg_id)
+
+    async def complete_pending_voice(
+        self, chat_id: str, voice_msg_id: str, status_msg_id: str
+    ) -> None:
+        """Mark a voice note as no longer pending."""
+        await self._pending_voice.complete(chat_id, voice_msg_id, status_msg_id)
+
+    async def handle(
+        self,
+        request: VoiceNoteRequest,
+        *,
+        message_handler: MessageHandler | None,
+        queue_send_message: QueueSend,
+        queue_delete_message: QueueDelete,
+    ) -> bool:
+        """Transcribe a voice note and hand the resulting turn to messaging."""
+        if await self.reply_if_disabled(request.reply_text):
+            return True
+
+        if message_handler is None:
+            return False
+
+        status_msg_id = await queue_send_message(
+            request.chat_id,
+            request.status_text,
+            reply_to=request.message_id,
+            parse_mode=request.status_parse_mode,
+            fire_and_forget=False,
+            message_thread_id=request.message_thread_id,
+        )
+        status_msg_id_text = str(status_msg_id)
+        await self.register_pending_voice(
+            request.chat_id,
+            request.message_id,
+            status_msg_id_text,
+        )
+        handed_off = False
+
+        with tempfile.NamedTemporaryFile(
+            suffix=request.temp_suffix, delete=False
+        ) as tmp:
+            tmp_path = Path(tmp.name)
+
+        try:
+            await request.download_to(tmp_path)
+
+            transcribed = await self._voice_transcription.transcribe(
+                tmp_path,
+                request.content_type,
+                whisper_model=self._whisper_model,
+                whisper_device=self._whisper_device,
+            )
+
+            if not await self.is_voice_still_pending(
+                request.chat_id,
+                request.message_id,
+            ):
+                await queue_delete_message(request.chat_id, status_msg_id_text)
+                return True
+
+            await self.complete_pending_voice(
+                request.chat_id,
+                request.message_id,
+                status_msg_id_text,
+            )
+            handed_off = True
+
+            incoming = IncomingMessage(
+                text=transcribed,
+                chat_id=request.chat_id,
+                user_id=request.user_id,
+                message_id=request.message_id,
+                platform=request.platform,
+                reply_to_message_id=request.reply_to_message_id,
+                message_thread_id=request.message_thread_id,
+                username=request.username,
+                raw_event=request.raw_event,
+                status_message_id=status_msg_id,
+            )
+
+            self._log_transcription(request, transcribed)
+            await message_handler(incoming)
+            return True
+        except ValueError as e:
+            await self._clear_failed_pending_voice(
+                request,
+                status_msg_id_text,
+                queue_delete_message,
+                handed_off=handed_off,
+            )
+            await request.reply_text(format_user_error_preview(e))
+            return True
+        except ImportError as e:
+            await self._clear_failed_pending_voice(
+                request,
+                status_msg_id_text,
+                queue_delete_message,
+                handed_off=handed_off,
+            )
+            await request.reply_text(format_user_error_preview(e))
+            return True
+        except Exception as e:
+            await self._clear_failed_pending_voice(
+                request,
+                status_msg_id_text,
+                queue_delete_message,
+                handed_off=handed_off,
+            )
+            if self._log_api_error_tracebacks:
+                logger.error("Voice transcription failed: {}", e)
+            else:
+                logger.error(
+                    "Voice transcription failed: exc_type={}",
+                    type(e).__name__,
+                )
+            await request.reply_text(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
+            return True
+        finally:
+            with contextlib.suppress(OSError):
+                tmp_path.unlink(missing_ok=True)
+
+    async def _clear_failed_pending_voice(
+        self,
+        request: VoiceNoteRequest,
+        status_msg_id: str,
+        queue_delete_message: QueueDelete,
+        *,
+        handed_off: bool,
+    ) -> None:
+        await self.complete_pending_voice(
+            request.chat_id,
+            request.message_id,
+            status_msg_id,
+        )
+        if not handed_off:
+            with contextlib.suppress(Exception):
+                await queue_delete_message(request.chat_id, status_msg_id)
+
+    def _log_transcription(self, request: VoiceNoteRequest, transcribed: str) -> None:
+        label = request.platform.upper()
+        if self._log_raw_messaging_content:
+            logger.info(
+                "{}_VOICE: chat_id={} message_id={} transcribed={!r}",
+                label,
+                request.chat_id,
+                request.message_id,
+                (transcribed[:80] + "..." if len(transcribed) > 80 else transcribed),
+            )
+        else:
+            logger.info(
+                "{}_VOICE: chat_id={} message_id={} transcribed_len={}",
+                label,
+                request.chat_id,
+                request.message_id,
+                len(transcribed),
+            )

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "free-claude-code"
-version = "2.3.7"
+version = "2.3.8"
 description = "Middleware between Claude Code CLI (Anthropic API) and NVIDIA NIM"
 readme = "README.md"
 requires-python = ">=3.14.0"

tests/contracts/test_import_boundaries.py

@@ -240,6 +240,24 @@ def test_messaging_workflow_uses_split_runtime_owners() -> None:
     assert offenders == []
 
 
+def test_messaging_platforms_use_shared_outbox_and_voice_flow() -> None:
+    repo_root = Path(__file__).resolve().parents[2]
+    platforms_root = repo_root / "messaging" / "platforms"
+
+    assert (platforms_root / "outbox.py").exists()
+    assert (platforms_root / "voice_flow.py").exists()
+
+    for adapter in {
+        platforms_root / "telegram.py",
+        platforms_root / "discord.py",
+    }:
+        text = adapter.read_text(encoding="utf-8")
+        assert "PlatformOutbox" in text
+        assert "VoiceNoteFlow" in text
+        assert "from ..voice" not in text
+        assert "NamedTemporaryFile" not in text
+
+
 def _imports_matching(
     roots: list[Path], *, forbidden_prefixes: tuple[str, ...]
 ) -> list[str]:

tests/messaging/test_platform_outbox.py

@@ -0,0 +1,108 @@
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from messaging.platforms.outbox import PlatformOutbox
+
+
+def _noop_outbox(*, limiter=None) -> PlatformOutbox:
+    async def send(
+        chat_id: str,
+        text: str,
+        reply_to: str | None,
+        parse_mode: str | None,
+        message_thread_id: str | None,
+    ) -> str:
+        return f"{chat_id}:{text}:{reply_to}:{parse_mode}:{message_thread_id}"
+
+    async def edit(
+        chat_id: str,
+        message_id: str,
+        text: str,
+        parse_mode: str | None,
+    ) -> None:
+        return None
+
+    async def delete(chat_id: str, message_id: str) -> None:
+        return None
+
+    async def delete_many(chat_id: str, message_ids: list[str]) -> None:
+        return None
+
+    return PlatformOutbox(
+        get_limiter=lambda: limiter,
+        send=send,
+        edit=edit,
+        delete=delete,
+        delete_many=delete_many,
+    )
+
+
+@pytest.mark.asyncio
+async def test_queue_send_without_limiter_calls_raw_send() -> None:
+    outbox = _noop_outbox()
+
+    result = await outbox.queue_send_message(
+        "chat",
+        "hello",
+        reply_to="reply",
+        parse_mode="MarkdownV2",
+        fire_and_forget=False,
+        message_thread_id="thread",
+    )
+
+    assert result == "chat:hello:reply:MarkdownV2:thread"
+
+
+@pytest.mark.asyncio
+async def test_queue_edit_awaits_limiter_with_dedup_key() -> None:
+    limiter = MagicMock()
+    limiter.enqueue = AsyncMock()
+    outbox = _noop_outbox(limiter=limiter)
+
+    await outbox.queue_edit_message(
+        "chat",
+        "message",
+        "updated",
+        parse_mode="MarkdownV2",
+        fire_and_forget=False,
+    )
+
+    limiter.enqueue.assert_awaited_once()
+    operation = limiter.enqueue.call_args.args[0]
+    assert limiter.enqueue.call_args.kwargs["dedup_key"] == "edit:chat:message"
+    await operation()
+
+
+@pytest.mark.asyncio
+async def test_queue_delete_uses_fire_and_forget_with_dedup_key() -> None:
+    limiter = MagicMock()
+    outbox = _noop_outbox(limiter=limiter)
+
+    await outbox.queue_delete_message("chat", "message", fire_and_forget=True)
+
+    limiter.fire_and_forget.assert_called_once()
+    assert limiter.fire_and_forget.call_args.kwargs["dedup_key"] == "del:chat:message"
+
+
+@pytest.mark.asyncio
+async def test_queue_delete_many_skips_empty_batches() -> None:
+    limiter = MagicMock()
+    outbox = _noop_outbox(limiter=limiter)
+
+    await outbox.queue_delete_messages("chat", [], fire_and_forget=True)
+
+    limiter.fire_and_forget.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_queue_delete_many_dedups_by_batch() -> None:
+    limiter = MagicMock()
+    outbox = _noop_outbox(limiter=limiter)
+
+    await outbox.queue_delete_messages("chat", ["1", "2"], fire_and_forget=True)
+
+    limiter.fire_and_forget.assert_called_once()
+    assert limiter.fire_and_forget.call_args.kwargs["dedup_key"] == (
+        f"del_bulk:chat:{hash(('1', '2'))}"
+    )

tests/messaging/test_platform_voice_flow.py

@@ -0,0 +1,244 @@
+from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
+
+from messaging.platforms.voice_flow import (
+    VOICE_DISABLED_MESSAGE,
+    VOICE_TRANSCRIPTION_ERROR_MESSAGE,
+    VoiceNoteFlow,
+    VoiceNoteRequest,
+    audio_suffix_from_metadata,
+    is_audio_metadata,
+)
+
+
+def _flow(*, enabled: bool = True) -> VoiceNoteFlow:
+    return VoiceNoteFlow(
+        voice_note_enabled=enabled,
+        whisper_model="base",
+        whisper_device="cpu",
+        hf_token="",
+        nvidia_nim_api_key="",
+        log_raw_messaging_content=False,
+        log_api_error_tracebacks=False,
+    )
+
+
+def _request(
+    *,
+    download_to=None,
+    reply_text=None,
+    message_id: str = "voice",
+) -> VoiceNoteRequest:
+    async def default_download_to(path: Path) -> None:
+        path.write_bytes(b"voice")
+
+    return VoiceNoteRequest(
+        platform="telegram",
+        chat_id="chat",
+        user_id="user",
+        message_id=message_id,
+        raw_event={"raw": True},
+        content_type="audio/ogg",
+        temp_suffix=".ogg",
+        status_text="transcribing",
+        status_parse_mode="MarkdownV2",
+        message_thread_id="thread",
+        reply_to_message_id="reply",
+        download_to=download_to or default_download_to,
+        reply_text=reply_text or AsyncMock(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_voice_flow_success_builds_incoming_message(monkeypatch) -> None:
+    flow = _flow()
+    transcribe = AsyncMock(return_value="hello from voice")
+    monkeypatch.setattr(flow._voice_transcription, "transcribe", transcribe)
+    handler = AsyncMock()
+    queue_send = AsyncMock(return_value="status")
+    queue_delete = AsyncMock()
+    downloaded_paths: list[Path] = []
+
+    async def download_to(path: Path) -> None:
+        downloaded_paths.append(path)
+        path.write_bytes(b"voice")
+
+    handled = await flow.handle(
+        _request(download_to=download_to),
+        message_handler=handler,
+        queue_send_message=queue_send,
+        queue_delete_message=queue_delete,
+    )
+
+    assert handled is True
+    queue_send.assert_awaited_once_with(
+        "chat",
+        "transcribing",
+        reply_to="voice",
+        parse_mode="MarkdownV2",
+        fire_and_forget=False,
+        message_thread_id="thread",
+    )
+    queue_delete.assert_not_awaited()
+    handler.assert_awaited_once()
+    incoming = handler.call_args.args[0]
+    assert incoming.text == "hello from voice"
+    assert incoming.chat_id == "chat"
+    assert incoming.user_id == "user"
+    assert incoming.message_id == "voice"
+    assert incoming.reply_to_message_id == "reply"
+    assert incoming.message_thread_id == "thread"
+    assert incoming.status_message_id == "status"
+    assert downloaded_paths and not downloaded_paths[0].exists()
+
+
+@pytest.mark.asyncio
+async def test_voice_flow_disabled_replies_without_transcribing(monkeypatch) -> None:
+    flow = _flow(enabled=False)
+    transcribe = AsyncMock(return_value="should not run")
+    monkeypatch.setattr(flow._voice_transcription, "transcribe", transcribe)
+    reply_text = AsyncMock()
+
+    handled = await flow.handle(
+        _request(reply_text=reply_text),
+        message_handler=AsyncMock(),
+        queue_send_message=AsyncMock(),
+        queue_delete_message=AsyncMock(),
+    )
+
+    assert handled is True
+    reply_text.assert_awaited_once_with(VOICE_DISABLED_MESSAGE)
+    transcribe.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_voice_flow_cancelled_transcription_deletes_status(monkeypatch) -> None:
+    flow = _flow()
+
+    async def canceling_transcribe(*args, **kwargs) -> str:
+        await flow.cancel_pending_voice("chat", "voice")
+        return "ignored"
+
+    monkeypatch.setattr(
+        flow._voice_transcription,
+        "transcribe",
+        AsyncMock(side_effect=canceling_transcribe),
+    )
+    handler = AsyncMock()
+    queue_send = AsyncMock(return_value="status")
+    queue_delete = AsyncMock()
+
+    handled = await flow.handle(
+        _request(),
+        message_handler=handler,
+        queue_send_message=queue_send,
+        queue_delete_message=queue_delete,
+    )
+
+    assert handled is True
+    handler.assert_not_awaited()
+    queue_delete.assert_awaited_once_with("chat", "status")
+
+
+@pytest.mark.asyncio
+async def test_voice_flow_download_failure_cleans_pending_state(monkeypatch) -> None:
+    flow = _flow()
+    transcribe = AsyncMock(return_value="should not run")
+    monkeypatch.setattr(flow._voice_transcription, "transcribe", transcribe)
+    reply_text = AsyncMock()
+    queue_delete = AsyncMock()
+
+    async def failing_download(_path: Path) -> None:
+        raise RuntimeError("download failed")
+
+    handled = await flow.handle(
+        _request(download_to=failing_download, reply_text=reply_text),
+        message_handler=AsyncMock(),
+        queue_send_message=AsyncMock(return_value="status"),
+        queue_delete_message=queue_delete,
+    )
+
+    assert handled is True
+    transcribe.assert_not_awaited()
+    queue_delete.assert_awaited_once_with("chat", "status")
+    reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
+    assert await flow.cancel_pending_voice("chat", "voice") is None
+
+
+@pytest.mark.asyncio
+async def test_voice_flow_transcription_failure_cleans_pending_state(
+    monkeypatch,
+) -> None:
+    flow = _flow()
+    monkeypatch.setattr(
+        flow._voice_transcription,
+        "transcribe",
+        AsyncMock(side_effect=RuntimeError("transcription failed")),
+    )
+    reply_text = AsyncMock()
+    queue_delete = AsyncMock()
+
+    handled = await flow.handle(
+        _request(reply_text=reply_text),
+        message_handler=AsyncMock(),
+        queue_send_message=AsyncMock(return_value="status"),
+        queue_delete_message=queue_delete,
+    )
+
+    assert handled is True
+    queue_delete.assert_awaited_once_with("chat", "status")
+    reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
+    assert await flow.cancel_pending_voice("chat", "voice") is None
+
+
+@pytest.mark.asyncio
+async def test_voice_flow_handler_failure_cleans_pending_without_deleting_status(
+    monkeypatch,
+) -> None:
+    flow = _flow()
+    monkeypatch.setattr(
+        flow._voice_transcription,
+        "transcribe",
+        AsyncMock(return_value="hello from voice"),
+    )
+    reply_text = AsyncMock()
+    queue_delete = AsyncMock()
+
+    async def failing_handler(_incoming) -> None:
+        raise RuntimeError("handler failed")
+
+    handled = await flow.handle(
+        _request(reply_text=reply_text),
+        message_handler=failing_handler,
+        queue_send_message=AsyncMock(return_value="status"),
+        queue_delete_message=queue_delete,
+    )
+
+    assert handled is True
+    queue_delete.assert_not_awaited()
+    reply_text.assert_awaited_once_with(VOICE_TRANSCRIPTION_ERROR_MESSAGE)
+    assert await flow.cancel_pending_voice("chat", "voice") is None
+
+
+def test_audio_metadata_helpers() -> None:
+    assert is_audio_metadata("voice.ogg", "application/octet-stream") is True
+    assert is_audio_metadata("file.txt", "audio/ogg") is True
+    assert is_audio_metadata("file.txt", "text/plain") is False
+    assert (
+        audio_suffix_from_metadata(filename="voice.ogg", content_type="audio/mp4")
+        == ".mp4"
+    )
+    assert (
+        audio_suffix_from_metadata(filename="clip.m4a", content_type="audio/mp4")
+        == ".m4a"
+    )
+    assert (
+        audio_suffix_from_metadata(filename="clip.m4a", content_type="audio/mpeg")
+        == ".mp3"
+    )
+    assert audio_suffix_from_metadata(content_type="audio/mpeg") == ".mp3"
+    assert audio_suffix_from_metadata(filename="clip.m4a") == ".m4a"
+    assert audio_suffix_from_metadata(content_type="audio/mp4") == ".mp4"
+    assert audio_suffix_from_metadata(content_type="audio/wav") == ".wav"

uv.lock

@@ -561,7 +561,7 @@ wheels = [
 
 [[package]]
 name = "free-claude-code"
-version = "2.3.7"
+version = "2.3.8"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },