feat(telegram): support audio and video input

src/bub/builtin/hook_impl.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 from difflib import get_close_matches
 from pathlib import Path
-from typing import cast
+from typing import Any, cast
 
 import typer
 from loguru import logger
@@ -15,7 +15,7 @@
 from bub.channels.admission import AdmitDecision, SteeringInbox, TurnSnapshot
 from bub.channels.base import Channel
 from bub.channels.contracts import MessageHandler
-from bub.channels.message import ChannelMessage, MediaItem
+from bub.channels.message import ChannelMessage, MediaItem, audio_format_from_mime_type
 from bub.envelope import Envelope, content_of, field_of
 from bub.framework import BubFramework
 from bub.hooks import hookimpl
@@ -63,6 +63,16 @@
 DEFAULT_CONTINUE_PROMPT = "Continue the task until all targets are completed."
 
 
+def _input_audio_part(data_url: str, mime_type: str) -> dict[str, Any] | None:
+    prefix, separator, data = data_url.partition("base64,")
+    if not separator or not prefix.startswith("data:audio/") or not data:
+        return None
+    return {
+        "type": "input_audio",
+        "input_audio": {"data": data, "format": audio_format_from_mime_type(mime_type)},
+    }
+
+
 class BuiltinImpl:
     """Default hook implementations for basic runtime operations."""
 
@@ -198,13 +208,18 @@ async def build_prompt(self, message: ChannelMessage, session_id: str, state: Tu
         media_parts: list[dict] = []
         for item in cast("list[MediaItem]", media):
             match item.type:
-                case "image":
+                case "image" | "video":
                     data_url = await item.get_url()
                     if not data_url:
                         continue
-                    media_parts.append({"type": "image_url", "image_url": {"url": data_url}})
+                    part_type = f"{item.type}_url"
+                    media_parts.append({"type": part_type, part_type: {"url": data_url}})
+                case "audio":
+                    data_url = await item.get_url()
+                    if data_url and (audio_part := _input_audio_part(data_url, item.mime_type)):
+                        media_parts.append(audio_part)
                 case _:
-                    pass  # TODO: Not supported for now
+                    pass
         if media_parts:
             return [{"type": "text", "text": text}, *media_parts]
         return text

src/bub/builtin/model_runner.py

@@ -28,6 +28,7 @@
 
 from bub.builtin.codex_provider import OpenaiCodexProvider, should_use_openai_codex_provider
 from bub.builtin.settings import AgentSettings, ModelCandidate
+from bub.channels.message import audio_mime_type_from_format
 from bub.errors import BubError, ErrorKind
 from bub.hooks.interception import (
     AgentHooks,
@@ -45,6 +46,7 @@
 )
 TOOL_ARGUMENTS_ADAPTER = TypeAdapter(dict[str, Any])
 CompletionResult = ChatCompletion | ParsedChatCompletion[Any] | AsyncIterator[ChatCompletionChunk]
+GOOGLE_FILE_CONTENT_PROVIDERS = frozenset({LLMProvider.GEMINI, LLMProvider.VERTEXAI})
 
 
 def _extra_options(llm: AnyLLM, *, stream: bool) -> dict[str, Any]:
@@ -56,6 +58,48 @@ def _extra_options(llm: AnyLLM, *, stream: bool) -> dict[str, Any]:
     return {}
 
 
+def _adapt_messages_for_provider(messages: list[dict[str, Any]], provider: LLMProvider) -> list[dict[str, Any]]:
+    """Translate canonical multimodal blocks when a provider uses a different wire format."""
+    if provider not in GOOGLE_FILE_CONTENT_PROVIDERS:
+        return messages
+
+    adapted_messages: list[dict[str, Any]] = []
+    for message in messages:
+        content = message.get("content")
+        if not isinstance(content, list):
+            adapted_messages.append(message)
+            continue
+
+        adapted_content: list[Any] = []
+        changed = False
+        for part in content:
+            if not isinstance(part, dict):
+                adapted_content.append(part)
+                continue
+
+            if part.get("type") == "video_url":
+                video_url = part.get("video_url")
+                url = video_url.get("url") if isinstance(video_url, dict) else None
+                if isinstance(url, str) and url:
+                    adapted_content.append({"type": "file", "file": {"file_data": url}})
+                    changed = True
+                    continue
+            elif part.get("type") == "input_audio":
+                input_audio = part.get("input_audio")
+                data = input_audio.get("data") if isinstance(input_audio, dict) else None
+                audio_format = input_audio.get("format") if isinstance(input_audio, dict) else None
+                if isinstance(data, str) and data and isinstance(audio_format, str) and audio_format:
+                    mime_type = audio_mime_type_from_format(audio_format)
+                    file_data = f"data:{mime_type};base64,{data}"
+                    adapted_content.append({"type": "file", "file": {"file_data": file_data}})
+                    changed = True
+                    continue
+            adapted_content.append(part)
+
+        adapted_messages.append({**message, "content": adapted_content} if changed else message)
+    return adapted_messages
+
+
 class ModelRunner:
     def __init__(self, settings: AgentSettings, hooks: AgentHooks | None = None) -> None:
         self.settings = settings
@@ -90,12 +134,12 @@ async def completion_response(
         reasoning_effort: str | None = None,
     ) -> CompletionResult:
         tool_payloads = [tool.to_schema() for tool in tools] or None
-        completion_messages: list[dict[str, Any] | ChatCompletionMessage] = list(messages)
         clients = list(self.iter_llm_clients(model))
         completion_error: Exception | None = None
         for index, (candidate, llm) in enumerate(clients):
             try:
                 streaming = llm.SUPPORTS_COMPLETION_STREAMING
+                completion_messages = _adapt_messages_for_provider(messages, candidate.provider)
                 completion_kwargs = {
                     **self.settings.completion_args,
                     **_extra_options(llm, stream=streaming),

src/bub/channels/message.py

@@ -9,6 +9,32 @@
 type MessageKind = Literal["error", "normal", "command"]
 type MediaType = Literal["image", "audio", "video", "document"]
 
+_AUDIO_FORMAT_TO_MIME_TYPE = {
+    "aiff": "audio/aiff",
+    "flac": "audio/flac",
+    "m4a": "audio/mp4",
+    "mp3": "audio/mpeg",
+    "ogg": "audio/ogg",
+    "wav": "audio/wav",
+    "webm": "audio/webm",
+}
+_AUDIO_MIME_TYPE_TO_FORMAT = {
+    **{mime_type: audio_format for audio_format, mime_type in _AUDIO_FORMAT_TO_MIME_TYPE.items()},
+    "audio/x-aiff": "aiff",
+    "audio/x-flac": "flac",
+    "audio/x-m4a": "m4a",
+    "audio/x-wav": "wav",
+}
+
+
+def audio_format_from_mime_type(mime_type: str) -> str:
+    normalized = mime_type.partition(";")[0].strip().lower()
+    return _AUDIO_MIME_TYPE_TO_FORMAT.get(normalized, normalized.removeprefix("audio/") or "unknown")
+
+
+def audio_mime_type_from_format(audio_format: str) -> str:
+    return _AUDIO_FORMAT_TO_MIME_TYPE.get(audio_format, f"audio/{audio_format}")
+
 
 @dataclass
 class MediaItem:
@@ -18,14 +44,16 @@ class MediaItem:
     mime_type: str
     filename: str | None = None
     url: str | None = None
-    data_fetcher: Callable[[], Awaitable[bytes]] | None = None
+    data_fetcher: Callable[[], Awaitable[bytes | None]] | None = None
 
     async def get_url(self) -> str | None:
         """Get a URL for the media, fetching data if necessary."""
         if self.url:
             return self.url
         if self.data_fetcher is not None:
             data = await self.data_fetcher()
+            if data is None:
+                return None
             return f"data:{self.mime_type};base64,{base64.b64encode(data).decode('utf-8')}"
         return None
 

src/bub/channels/telegram.py

@@ -378,7 +378,7 @@ async def _parse_audio(self, message: Message) -> tuple[str, dict[str, Any] | No
         duration = audio.duration or 0
         metadata = exclude_none({
             "file_id": audio.file_id,
-            "mime_type": audio.mime_type,
+            "mime_type": audio.mime_type or "audio/mpeg",
             "file_size": audio.file_size,
             "duration": audio.duration,
             "title": audio.title,
@@ -389,12 +389,12 @@ async def _parse_audio(self, message: Message) -> tuple[str, dict[str, Any] | No
             return f"[Audio: {performer} - {title} ({duration}s)]", metadata
         return f"[Audio: {title} ({duration}s)]", metadata
 
-    async def _download_media(self, file_id: str, file_size: int) -> bytes | None:
+    async def _download_media(self, file_id: str, file_size: int | None) -> bytes | None:
         if not file_id:
             raise ValueError("file_id must not be empty")
         if self._bot_getter is None:
             raise RuntimeError("Telegram bot is not configured for media downloads.")
-        if file_size > 2 * 1024 * 1024:  # limit to 2MB
+        if file_size is not None and file_size > 2 * 1024 * 1024:  # limit to 2MB
             return None
         bot = self._bot_getter()
         if bot is None:
@@ -441,7 +441,7 @@ async def _parse_video(self, message: Message) -> tuple[str, dict[str, Any] | No
             "width": video.width,
             "height": video.height,
             "duration": video.duration,
-            "mime_type": video.mime_type,
+            "mime_type": video.mime_type or "video/mp4",
             "data_fetcher": lambda: self._download_media(video.file_id, video.file_size),
         })
         return formatted, metadata

tests/test_builtin_model_runner.py

@@ -10,12 +10,54 @@
 from any_llm.providers.openai.base import BaseOpenAIProvider
 from any_llm.types.completion import ChatCompletionChunk, ChatCompletionMessageFunctionToolCall, Function
 
-from bub.builtin.model_runner import ModelRunner, tool_invocation_from_native
+from bub.builtin.model_runner import ModelRunner, _adapt_messages_for_provider, tool_invocation_from_native
 from bub.builtin.settings import AgentSettings, ModelCandidate
 from bub.tape import AsyncTapeStoreAdapter, InMemoryTapeStore, Tape, TapeContext
 from bub.tools import ToolExecutor
 
 
+@pytest.mark.parametrize("provider", [LLMProvider.GEMINI, LLMProvider.VERTEXAI])
+def test_adapt_messages_converts_video_url_for_google_providers(provider: LLMProvider) -> None:
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "describe this video"},
+                {"type": "video_url", "video_url": {"url": "data:video/mp4;base64,dmlkZW8="}},
+                {"type": "input_audio", "input_audio": {"data": "YXVkaW8=", "format": "ogg"}},
+            ],
+        }
+    ]
+
+    result = _adapt_messages_for_provider(messages, provider)
+
+    assert result == [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "describe this video"},
+                {"type": "file", "file": {"file_data": "data:video/mp4;base64,dmlkZW8="}},
+                {"type": "file", "file": {"file_data": "data:audio/ogg;base64,YXVkaW8="}},
+            ],
+        }
+    ]
+    assert messages[0]["content"][1]["type"] == "video_url"
+
+
+def test_adapt_messages_keeps_native_multimodal_blocks_for_openrouter() -> None:
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "video_url", "video_url": {"url": "data:video/mp4;base64,dmlkZW8="}},
+                {"type": "input_audio", "input_audio": {"data": "YXVkaW8=", "format": "ogg"}},
+            ],
+        }
+    ]
+
+    assert _adapt_messages_for_provider(messages, LLMProvider.OPENROUTER) is messages
+
+
 @pytest.mark.asyncio
 async def test_unknown_tool_placeholder_surfaces_error_without_hooks() -> None:
     tool_call = ChatCompletionMessageFunctionToolCall(