diff --git a/contributors/emails/mkoduri73@gmail.com b/contributors/emails/mkoduri73@gmail.com
new file mode 100644
index 00000000000..8bca76a58e2
--- /dev/null
+++ b/contributors/emails/mkoduri73@gmail.com
@@ -0,0 +1,2 @@
+MaheshBhushan
+# PR #66492 salvage (#66377)
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 7f451f8c4c0..d91f2a79982 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -17,6 +17,7 @@ import os
 import html as _html
 import re
 import threading
+import time
 from contextvars import ContextVar
 from datetime import datetime, timezone
 from typing import Dict, List, Optional, Set, Any
@@ -543,6 +544,24 @@ _UPDATER_STOP_TIMEOUT = 15.0
 # reconnect ladder from stalling indefinitely and allows the heartbeat loop to
 # trigger its own recovery path. Refs: NousResearch/hermes-agent#59614
 _UPDATER_START_TIMEOUT = 30.0
+# shutdown()/initialize() on the getUpdates httpx request close and rebuild the
+# connection pool. When a connection is wedged on a stale CLOSE-WAIT socket that
+# close can block forever, hanging _drain_polling_connections() and freezing the
+# whole reconnect ladder (the tracked _polling_error_task never completes, so
+# every escalation path stays gated behind its in-flight guard). Bound the drain
+# so the ladder always advances toward the fatal-restart escalation. Matches
+# _UPDATER_STOP_TIMEOUT. Refs: NousResearch/hermes-agent#66377
+_DRAIN_TIMEOUT = 15.0
+# Cause-agnostic wedged-recovery watchdog (#66377). Every recovery path (the
+# reconnect ladder's re-entry, the pending-update probe, PTB's error callback)
+# gates new recovery on ``_polling_error_task.done()``; if that task ever wedges
+# on a hung await that no local bound covers, the whole gateway goes silently
+# deaf with nothing retrying. The heartbeat loop force-escalates a recovery task
+# that stays in-flight far longer than any healthy ladder attempt could take —
+# stop (_UPDATER_STOP_TIMEOUT) + drain (2x_DRAIN_TIMEOUT) + start
+# (_UPDATER_START_TIMEOUT) + max backoff (60s) is ~135s, so 300s is
+# unambiguously stuck.
+_POLLING_ERROR_TASK_STUCK_TIMEOUT = 300.0
 # A generation is not healthy until the dedicated getUpdates request returns
 # successfully. This exceeds a normal long-poll cycle for healthy idle bots.
 _POLLING_PROGRESS_TIMEOUT = 60.0
@@ -1987,20 +2006,22 @@ class TelegramAdapter(BasePlatformAdapter):
         except Exception:
             return
         try:
-            await polling_req.shutdown()
+            # Bounded: a wedged CLOSE-WAIT socket can make this close hang
+            # forever and freeze the reconnect ladder (#66377).
+            await asyncio.wait_for(polling_req.shutdown(), timeout=_DRAIN_TIMEOUT)
         except Exception:
             logger.debug(
-                "[%s] Polling request shutdown failed (non-fatal)",
+                "[%s] Polling request shutdown failed/timed out (non-fatal)",
                 self.name, exc_info=True,
             )
         try:
-            await polling_req.initialize()
+            await asyncio.wait_for(polling_req.initialize(), timeout=_DRAIN_TIMEOUT)
             logger.debug(
                 "[%s] Polling request pool drained before reconnect", self.name
             )
         except Exception:
             logger.debug(
-                "[%s] Polling request re-initialize failed (non-fatal)",
+                "[%s] Polling request re-initialize failed/timed out (non-fatal)",
                 self.name, exc_info=True,
             )
 
@@ -2454,6 +2475,16 @@ class TelegramAdapter(BasePlatformAdapter):
         HEARTBEAT_INTERVAL = 90   # seconds between probes
         PROBE_TIMEOUT = 15        # seconds before declaring the path dead
 
+        # Wedged-recovery watchdog state (#66377). Tracked locally so no
+        # _polling_error_task assignment site needs to stamp a timestamp: the
+        # heartbeat notes when it first observes a given recovery task still
+        # in-flight, and force-escalates if the *same* task object is still
+        # running after _POLLING_ERROR_TASK_STUCK_TIMEOUT. A healthy ladder
+        # attempt completes (task done) or chains to a new task well before
+        # then, so a single long-lived task is unambiguously wedged.
+        stuck_task_ref: Optional[asyncio.Task] = None
+        stuck_task_since = 0.0
+
         while True:
             try:
                 await asyncio.sleep(HEARTBEAT_INTERVAL)
@@ -2461,6 +2492,42 @@ class TelegramAdapter(BasePlatformAdapter):
                     return
                 if self.has_fatal_error:
                     return
+
+                # Independent wedged-recovery watchdog (#66377): if the tracked
+                # recovery task has hung (any await no local bound covers), every
+                # other recovery path is gated behind it and returns early
+                # forever — the gateway stays alive but deaf. Force a
+                # retryable-fatal so the background reconnector rebuilds the
+                # adapter instead of relying on the frozen ladder.
+                recovery_task = self._polling_error_task
+                if recovery_task is not None and not recovery_task.done():
+                    now = time.monotonic()
+                    if recovery_task is not stuck_task_ref:
+                        stuck_task_ref = recovery_task
+                        stuck_task_since = now
+                    elif now - stuck_task_since > _POLLING_ERROR_TASK_STUCK_TIMEOUT:
+                        stuck_for = now - stuck_task_since
+                        logger.error(
+                            "[%s] Telegram reconnect task wedged for %.0fs with no "
+                            "ladder progress; forcing retryable-fatal so the gateway "
+                            "reconnects instead of staying silently deaf.",
+                            self.name, stuck_for,
+                        )
+                        try:
+                            recovery_task.cancel()
+                        except Exception:
+                            pass
+                        self._set_fatal_error(
+                            "telegram_network_error",
+                            "Telegram reconnect task wedged for %.0fs; forcing "
+                            "gateway reconnect." % stuck_for,
+                            retryable=True,
+                        )
+                        await self._notify_fatal_error()
+                        return
+                else:
+                    stuck_task_ref = None
+
                 bot = self._app.bot if self._app else None
                 if bot is None:
                     continue
diff --git a/tests/gateway/test_telegram_network_reconnect.py b/tests/gateway/test_telegram_network_reconnect.py
index c1c10726755..4ffa4a99dd8 100644
--- a/tests/gateway/test_telegram_network_reconnect.py
+++ b/tests/gateway/test_telegram_network_reconnect.py
@@ -325,6 +325,104 @@ async def test_initialize_still_runs_when_shutdown_fails():
     mock_app.updater.start_polling.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_reconnect_continues_if_drain_hangs(monkeypatch):
+    """If the polling request drain HANGS (wedged httpx pool close on a
+    CLOSE-WAIT socket), the reconnect ladder must still advance rather than
+    freezing the tracked _polling_error_task forever.
+
+    Regression test for #66377: an unbounded ``shutdown()`` /
+    ``initialize()`` in ``_drain_polling_connections`` leaves the handler
+    task pending, which gates every escalation path and silently kills the
+    gateway. The drain awaits are bounded by ``_DRAIN_TIMEOUT``, so the
+    handler must complete and reach ``start_polling`` within a hard bound.
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 1
+
+    mock_app, mock_polling_req = _make_mock_app()
+
+    async def _hang(*args, **kwargs):
+        await asyncio.Event().wait()  # never returns
+
+    # Both drain awaits wedge indefinitely.
+    mock_polling_req.shutdown = AsyncMock(side_effect=_hang)
+    mock_polling_req.initialize = AsyncMock(side_effect=_hang)
+    adapter._app = mock_app
+
+    # Keep the drain timeout tiny so the test stays fast; the real default
+    # is generous enough not to truncate healthy closes.
+    monkeypatch.setattr(tg_adapter, "_DRAIN_TIMEOUT", 0.01, raising=False)
+
+    with patch("asyncio.sleep", new_callable=AsyncMock):
+        # Hard outer bound: on unfixed code the drain hangs forever and this
+        # trips; with the fix the inner wait_for releases well before it.
+        await asyncio.wait_for(
+            adapter._handle_polling_network_error(Exception("Timed out")),
+            timeout=5,
+        )
+
+    # Ladder advanced past the wedged drain despite it never returning.
+    mock_app.updater.start_polling.assert_called_once()
+    assert adapter._polling_network_error_count == 2
+    # The tracked task must not be stuck pending — otherwise every
+    # escalation path stays gated behind an in-flight guard.
+    assert (
+        adapter._polling_error_task is None
+        or adapter._polling_error_task.done()
+    )
+
+
+@pytest.mark.asyncio
+async def test_heartbeat_force_escalates_wedged_recovery_task(monkeypatch):
+    """#66377: the heartbeat is an independent, cause-agnostic watchdog.
+
+    Every recovery path (ladder re-entry, pending-update probe, PTB error
+    callback) gates new recovery on ``_polling_error_task.done()``. If that task
+    wedges on ANY hung await — not just the drain closed by #66492 — the gateway
+    stays alive but deaf with nothing retrying. The heartbeat must detect a
+    recovery task that stays in-flight past ``_POLLING_ERROR_TASK_STUCK_TIMEOUT``
+    and force a retryable-fatal so the background reconnector rebuilds the
+    adapter.
+    """
+    adapter = _make_adapter()
+
+    async def _wedged():
+        await asyncio.Event().wait()  # never completes — simulates the hang
+
+    wedged_task = asyncio.ensure_future(_wedged())
+    adapter._polling_error_task = wedged_task
+
+    mock_bot = MagicMock()
+    mock_bot.get_me = AsyncMock()
+    mock_app = MagicMock()
+    mock_app.bot = mock_bot
+    adapter._app = mock_app
+    adapter._probe_pending_updates = AsyncMock()
+    adapter._notify_fatal_error = AsyncMock()
+
+    # Controllable monotonic clock advanced by each (mocked) heartbeat sleep so
+    # the same wedged task is observed across the stuck threshold deterministically.
+    clock = [1000.0]
+
+    async def _fake_sleep(*_a, **_k):
+        clock[0] += 200.0
+
+    monkeypatch.setattr(tg_adapter.time, "monotonic", lambda: clock[0])
+
+    with patch("asyncio.sleep", new=AsyncMock(side_effect=_fake_sleep)):
+        await asyncio.wait_for(adapter._polling_heartbeat_loop(), timeout=5)
+
+    assert adapter.has_fatal_error, "wedged recovery task must force a fatal escalation"
+    adapter._notify_fatal_error.assert_awaited()
+
+    wedged_task.cancel()
+    try:
+        await wedged_task
+    except asyncio.CancelledError:
+        pass
+
+
 @pytest.mark.asyncio
 async def test_conflict_retry_also_drains_polling_connections():
     """_handle_polling_conflict must also drain the polling pool on retry."""