diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d019004f..91063df0 100644 --- a/ARCHITECTURE.md +++ b/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. diff --git a/messaging/platforms/discord.py b/messaging/platforms/discord.py index 53aa7dc7..958441d1 100644 --- a/messaging/platforms/discord.py +++ b/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, diff --git a/messaging/platforms/outbox.py b/messaging/platforms/outbox.py new file mode 100644 index 00000000..ae2c3395 --- /dev/null +++ b/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) diff --git a/messaging/platforms/telegram.py b/messaging/platforms/telegram.py index a8f34dff..6fb62dd7 100644 --- a/messaging/platforms/telegram.py +++ b/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, + ) diff --git a/messaging/platforms/voice_flow.py b/messaging/platforms/voice_flow.py new file mode 100644 index 00000000..51e672e5 --- /dev/null +++ b/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), + ) diff --git a/pyproject.toml b/pyproject.toml index 3dd2783f..c039880e 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/contracts/test_import_boundaries.py b/tests/contracts/test_import_boundaries.py index 7cf6aa24..5eaf7092 100644 --- a/tests/contracts/test_import_boundaries.py +++ b/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]: diff --git a/tests/messaging/test_platform_outbox.py b/tests/messaging/test_platform_outbox.py new file mode 100644 index 00000000..b0f26ce0 --- /dev/null +++ b/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'))}" + ) diff --git a/tests/messaging/test_platform_voice_flow.py b/tests/messaging/test_platform_voice_flow.py new file mode 100644 index 00000000..334a210b --- /dev/null +++ b/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" diff --git a/uv.lock b/uv.lock index 16668a6f..cfa2734d 100644 --- a/uv.lock +++ b/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" },