Guard oversized Telegram video downloads

plugins/platforms/telegram/adapter.py

@@ -4707,8 +4707,7 @@ class TelegramAdapter(BasePlatformAdapter):
             size_text = "unknown size"
         return (
             f"[Telegram {label} skipped: file size {size_text} exceeds the "
-            f"{limit_mb} MB limit. Ask the user to send a shorter voice note "
-            "or a smaller audio file.]"
+            f"{limit_mb} MB limit. Ask the user to send a smaller file.]"
         )
 
     def _telegram_media_size_allowed(self, source: Any, label: str) -> tuple[bool, Optional[str]]:
@@ -6600,6 +6599,12 @@ class TelegramAdapter(BasePlatformAdapter):
 
         elif msg.video:
             try:
+                allowed, note = self._telegram_media_size_allowed(msg.video, "video file")
+                if not allowed:
+                    event.text = self._append_observed_note(event.text, note or "")
+                    logger.info("[Telegram] Skipped oversized user video (size=%s)", getattr(msg.video, "file_size", None))
+                    await self.handle_message(event)
+                    return
                 file_obj = await msg.video.get_file()
                 video_bytes = await file_obj.download_as_bytearray()
                 ext = ".mp4"

tests/gateway/test_telegram_voice_v0_regressions.py

@@ -48,6 +48,52 @@ def test_telegram_audio_size_gate_rejects_oversized_media_before_download():
     assert "voice message" in note
 
 
+@pytest.mark.asyncio
+async def test_telegram_video_size_gate_rejects_oversized_media_before_download():
+    adapter = object.__new__(TelegramAdapter)
+    adapter._max_doc_bytes = 1024
+    adapter._should_process_message = lambda _message: True
+    adapter._build_message_event = lambda _message, _type, update_id=None: SimpleNamespace(
+        text="caption",
+        media_urls=[],
+        media_types=[],
+    )
+    adapter._apply_telegram_group_observe_attribution = lambda event: event
+
+    handled = []
+
+    async def handle_message(event):
+        handled.append(event)
+
+    adapter.handle_message = handle_message
+
+    class OversizedVideo:
+        file_size = 2048
+
+        async def get_file(self):  # pragma: no cover - failure path assertion
+            pytest.fail("oversized videos must not be downloaded")
+
+    msg = SimpleNamespace(
+        caption=None,
+        sticker=None,
+        photo=None,
+        voice=None,
+        audio=None,
+        video=OversizedVideo(),
+        document=None,
+        media_group_id=None,
+    )
+    update = SimpleNamespace(message=msg, update_id=1)
+
+    await TelegramAdapter._handle_media_message(adapter, update, SimpleNamespace())
+
+    assert len(handled) == 1
+    assert handled[0].media_urls == []
+    assert handled[0].media_types == []
+    assert "video file" in handled[0].text
+    assert "exceeds" in handled[0].text
+
+
 @pytest.mark.asyncio
 async def test_voice_tts_is_explicit_audio_reply_opt_in():
     adapter = SimpleNamespace(