fix(telegram): surface failed media downloads to user and agent, not a silent empty turn (#53912)

When a Telegram attachment download/cache fails (typically a transient
httpx.ConnectError to Telegram's CDN), the except handler logged a warning
and fell through to handle_message() with empty media and no text — the user
thought the file was delivered, the agent saw a content-less turn with no
signal an attachment was attempted, and the only record was a buried log line.

Adds _surface_media_cache_failure(): replies to the user in Telegram so they
know to retry, and appends an agent-visible notice to event.text via the
existing _append_observed_note channel so the agent knows an attachment was
attempted and failed. No new event fields (structured-event refactor is out
of scope per #23045). Wired into all five cache-failure sites — photo, voice,
audio, video, document — since they shared the identical silent fall-through.

Bug 1 from #23045 (unsupported types routed as fake user messages) no longer
exists on main: the document handler now accepts any file type, so there is no
rejection branch to fix.

Closes #23045

plugins/platforms/telegram/adapter.py

@@ -6101,6 +6101,47 @@ class TelegramAdapter(BasePlatformAdapter):
             return note
         return f"{existing}\n\n{note}"
 
+    async def _surface_media_cache_failure(
+        self,
+        msg: Message,
+        event: MessageEvent,
+        kind: str,
+        exc: Exception,
+        display_name: Optional[str] = None,
+    ) -> None:
+        """Surface a failed media download/cache on BOTH ends instead of swallowing it.
+
+        When download_as_bytearray()/cache_*_from_bytes() raises (typically a
+        transient httpx.ConnectError to Telegram's CDN), the attachment never
+        made it into event.media_urls. Without this, the handler falls through
+        and dispatches an empty turn: the user thinks the file was delivered,
+        the agent sees nothing, and the only record is a buried log warning.
+
+        This (1) replies to the user in Telegram so they know to retry, and
+        (2) appends an agent-visible notice to event.text via the existing
+        observed-note channel so the agent knows an attachment was attempted
+        and failed — never a silent empty turn. No new event fields (the
+        structured-event refactor is out of scope per #23045).
+        """
+        named = f" ({display_name})" if display_name else ""
+        try:
+            await msg.reply_text(
+                f"\u26a0\ufe0f Couldn't download your {kind}{named} "
+                f"({exc.__class__.__name__}). Please try sending it again."
+            )
+        except Exception as reply_err:
+            logger.warning(
+                "[Telegram] Failed to notify user about %s cache failure: %s",
+                kind,
+                reply_err,
+                exc_info=True,
+            )
+        agent_note = (
+            f"[The user attempted to send a {kind}{named} but it could not be "
+            f"downloaded ({exc.__class__.__name__}); they have been asked to retry.]"
+        )
+        event.text = self._append_observed_note(event.text, agent_note)
+
     def _observe_unmentioned_group_message(
         self,
         message: Message,
@@ -6562,6 +6603,7 @@ class TelegramAdapter(BasePlatformAdapter):
 
             except Exception as e:
                 logger.warning("[Telegram] Failed to cache photo: %s", e, exc_info=True)
+                await self._surface_media_cache_failure(msg, event, "photo", e)
 
         # Download voice/audio messages to cache for STT transcription
         if msg.voice:
@@ -6580,6 +6622,7 @@ class TelegramAdapter(BasePlatformAdapter):
                 logger.info("[Telegram] Cached user voice at %s", cached_path)
             except Exception as e:
                 logger.warning("[Telegram] Failed to cache voice: %s", e, exc_info=True)
+                await self._surface_media_cache_failure(msg, event, "voice message", e)
         elif msg.audio:
             try:
                 allowed, note = self._telegram_media_size_allowed(msg.audio, "audio file")
@@ -6596,6 +6639,7 @@ class TelegramAdapter(BasePlatformAdapter):
                 logger.info("[Telegram] Cached user audio at %s", cached_path)
             except Exception as e:
                 logger.warning("[Telegram] Failed to cache audio: %s", e, exc_info=True)
+                await self._surface_media_cache_failure(msg, event, "audio file", e)
 
         elif msg.video:
             try:
@@ -6619,6 +6663,7 @@ class TelegramAdapter(BasePlatformAdapter):
                 logger.info("[Telegram] Cached user video at %s", cached_path)
             except Exception as e:
                 logger.warning("[Telegram] Failed to cache video: %s", e, exc_info=True)
+                await self._surface_media_cache_failure(msg, event, "video file", e)
 
         # Download document files to cache for agent processing
         elif msg.document:
@@ -6751,6 +6796,10 @@ class TelegramAdapter(BasePlatformAdapter):
 
             except Exception as e:
                 logger.warning("[Telegram] Failed to cache document: %s", e, exc_info=True)
+                await self._surface_media_cache_failure(
+                    msg, event, "attachment", e,
+                    display_name=getattr(doc, "file_name", None) or None,
+                )
 
         media_group_id = getattr(msg, "media_group_id", None)
         if media_group_id:

tests/gateway/test_telegram_documents.py

@@ -106,6 +106,7 @@ def _make_message(document=None, caption=None, media_group_id=None, photo=None):
     msg.from_user.id = 1
     msg.from_user.full_name = "Test User"
     msg.message_thread_id = None
+    msg.reply_text = AsyncMock()
     return msg
 
 
@@ -397,7 +398,7 @@ class TestDocumentDownloadBlock:
 
     @pytest.mark.asyncio
     async def test_download_exception_handled(self, adapter):
-        """If get_file() raises, the handler logs the error without crashing."""
+        """If get_file() raises, the handler surfaces the failure without crashing."""
         doc = _make_document(file_name="crash.pdf", file_size=100)
         doc.get_file = AsyncMock(side_effect=RuntimeError("Telegram API down"))
         msg = _make_message(document=doc)
@@ -406,8 +407,73 @@ class TestDocumentDownloadBlock:
         # Should not raise
         await adapter._handle_media_message(update, MagicMock())
         # handle_message should still be called (the handler catches the exception)
+        # so the agent gets a turn — but now that turn carries a notice instead
+        # of being an empty/content-less event.
         adapter.handle_message.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_document_cache_failure_replies_and_signals_agent(self, adapter):
+        """A failed document download must surface on BOTH ends, not silently.
+
+        Regression for #23045 Bug 2: a CDN download/cache failure used to log a
+        warning and fall through to an empty agent turn — user thinks the file
+        arrived, agent sees nothing. Now the user gets a Telegram reply AND the
+        agent's event.text carries an attempted-attachment notice.
+        """
+        doc = _make_document(file_name="notes.md", mime_type="text/markdown", file_size=100)
+        doc.get_file = AsyncMock(side_effect=RuntimeError("Telegram CDN down"))
+        msg = _make_message(document=doc)
+        update = _make_update(msg)
+
+        await adapter._handle_media_message(update, MagicMock())
+
+        # 1. User is told the download failed, with the filename + exception type.
+        msg.reply_text.assert_awaited_once()
+        reply = msg.reply_text.await_args.args[0]
+        assert "Couldn't download" in reply
+        assert "notes.md" in reply
+        assert "RuntimeError" in reply
+
+        # 2. The agent still gets a turn, but event.text now carries a notice so
+        #    it knows an attachment was attempted and failed (not a silent empty turn).
+        adapter.handle_message.assert_called_once()
+        event = adapter.handle_message.call_args[0][0]
+        assert event.media_urls == []  # nothing cached
+        assert "could not be downloaded" in (event.text or "")
+        assert "notes.md" in (event.text or "")
+
+    @pytest.mark.asyncio
+    async def test_document_cache_failure_reply_error_is_nonfatal(self, adapter):
+        """If even the user-reply fails, the agent notice is still appended."""
+        doc = _make_document(file_name="x.bin", mime_type="application/octet-stream", file_size=100)
+        doc.get_file = AsyncMock(side_effect=RuntimeError("CDN down"))
+        msg = _make_message(document=doc)
+        msg.reply_text = AsyncMock(side_effect=RuntimeError("reply failed too"))
+        update = _make_update(msg)
+
+        # Must not raise despite reply_text blowing up
+        await adapter._handle_media_message(update, MagicMock())
+        adapter.handle_message.assert_called_once()
+        event = adapter.handle_message.call_args[0][0]
+        assert "could not be downloaded" in (event.text or "")
+
+    @pytest.mark.asyncio
+    async def test_voice_cache_failure_replies_and_signals_agent(self, adapter):
+        """Same fail-closed contract applies to the voice site (#23045 Bug 2 class)."""
+        msg = _make_message()
+        msg.voice = MagicMock()
+        msg.voice.file_size = 100
+        msg.voice.get_file = AsyncMock(side_effect=RuntimeError("CDN down"))
+        update = _make_update(msg)
+
+        await adapter._handle_media_message(update, MagicMock())
+
+        msg.reply_text.assert_awaited_once()
+        assert "voice message" in msg.reply_text.await_args.args[0]
+        adapter.handle_message.assert_called_once()
+        event = adapter.handle_message.call_args[0][0]
+        assert "could not be downloaded" in (event.text or "")
+
 
 class TestVideoDownloadBlock:
     @pytest.mark.asyncio