From dbc925b7550c2224631d6fb1a2fa0ec236d76beb Mon Sep 17 00:00:00 2001
From: r266-tech <r2668940489@gmail.com>
Date: Fri, 12 Jun 2026 04:48:54 +0800
Subject: [PATCH] Guard oversized Telegram video downloads

---
 plugins/platforms/telegram/adapter.py         |  9 +++-
 .../test_telegram_voice_v0_regressions.py     | 46 +++++++++++++++++++
 2 files changed, 53 insertions(+), 2 deletions(-)

diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index cf37684ee98..1f015c31485 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/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"
diff --git a/tests/gateway/test_telegram_voice_v0_regressions.py b/tests/gateway/test_telegram_voice_v0_regressions.py
index b7527601fbc..6e2143573c4 100644
--- a/tests/gateway/test_telegram_voice_v0_regressions.py
+++ b/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(