Telegram post-reconnect probe recovery bug fixes

NousResearch/hermes-agent · cluster 63243

Data as of 2026-10-22 12:15 UTC

Nothing open on GitHub. Fix merged in #64361. Check it covers your case.

open issues

open PRs

items grouped

Known?: Yes. 6 related items grouped here, 2 issues already closed.
Fix event: A fix was merged to main in PR #64361. Check it covers your case. No release data is tracked.
Tracker: The maintainers' tracker marks this partially resolved, deciding PR #64361. That is separate from the fix event.
Still open: Nothing open on GitHub.

Next step: read #64361, then check it covers your case.

Read #64361 Pick up a Lab task

These issues and pull requests address bugs in the Telegram integration's post-reconnect recovery logic, specifically around probe failures bypassing deduplication guards, misclassified auth errors, and serialization of recovery behavior after reconnection.

Open issues0

No open issues.

Open PRs0

No open pull requests.

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

#64361 fix(telegram): require getUpdates progress before polling is healthy

merged to main

files: plugins/platforms/telegram/adapter.py, scripts/release.py, tests/gateway/test_telegram_conflict.py, tests/gateway/test_telegram_network_reconnect.py, tests/gateway/test_telegram_polling_progress.py, tests/gateway/test_telegram_send_path_health.py, tests/test_telegram_polling_progress_ptb.py

Show the diff

plugins/platforms/telegram/adapter.py

@@ -17,6 +17,7 @@
 import html as _html
 import re
 import threading
+from contextvars import ContextVar
 from datetime import datetime, timezone
 from typing import Dict, List, Optional, Set, Any
 
@@ -474,6 +475,16 @@ def _rich_normalize_linebreaks(text: str) -> str:
 # 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
+# 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
+_POLLING_GENERATION_CONTEXT: ContextVar[Optional[int]] = ContextVar(
+    "telegram_polling_generation", default=None
+)
+
+
+class _PollingLifecycleAbort(RuntimeError):
+    """Internal control flow for polling startup fenced by teardown."""
 
 
 class TelegramAdapter(BasePlatformAdapter):
@@ -633,6 +644,11 @@ def __init__(self, config: PlatformConfig):
         self._polling_error_task: Optional[asyncio.Task] = None
         self._polling_conflict_count: int = 0
         self._polling_network_error_count: int = 0
+        self._polling_generation: int = 0
+        self._polling_progress_event = asyncio.Event()
+        self._polling_progress_accepting: bool = False
+        self._polling_progress_verifier_task: Optional[asyncio.Task] = None
+        self._polling_teardown_started: bool = False
         self._polling_error_callback_ref = None
         self._polling_heartbeat_task: Optional[asyncio.Task] = None
         # Consecutive heartbeat probes that saw queued updates the running
@@ -648,10 +664,9 @@ def __init__(self, config: PlatformConfig):
         # error_callback ever fires and the gateway silently stops receiving
         # messages with the process still alive (#55769).
         self._polling_not_running_count: int = 0
-        # After sustained reconnect storms the PTB httpx pool can return
-        # SendResult(success=True) for sends that never actually transmit.
-        # _handle_polling_network_error sets this; _verify_polling_after_reconnect
-        # clears it once getMe() confirms the Bot client is healthy.
+        # A polling generation stays degraded until the dedicated getUpdates
+        # request makes successful progress. start_polling() return and getMe()
+        # success on the general request path are not polling-health signals.
         # While True, send() short-circuits to a failure so callers
         # (cron live-adapter branch) fall through to standalone delivery.
         self._send_path_degraded: bool = False
@@ -1944,6 +1959,139 @@ async def _drain_polling_connections(self) -> None:
                 self.name, exc_info=True,
             )
 
+    def _begin_polling_generation(self) -> tuple[int, asyncio.Event]:
+        """Start accepting progress for a new getUpdates polling generation."""
+        if getattr(self, "_polling_teardown_started", False):
+            self._polling_progress_accepting = False
+            self._send_path_degraded = True
+            progress = getattr(self, "_polling_progress_event", None)
+            if progress is None:
+                progress = asyncio.Event()
+                self._polling_progress_event = progress
+            return getattr(self, "_polling_generation", 0), progress
+
+        verifier = getattr(self, "_polling_progress_verifier_task", None)
+        if verifier is not None and not verifier.done():
+            verifier.cancel()
+        self._polling_progress_verifier_task = None
+        self._polling_generation = getattr(self, "_polling_generation", 0) + 1
+        self._polling_progress_event = asyncio.Event()
+        self._polling_progress_accepting = True
+        self._send_path_degraded = True
+        return self._polling_generation, self._polling_progress_event
+
+    def _record_polling_progress(self, generation: int) -> None:
+        """Record successful getUpdates I/O for the current generation only."""
+        if getattr(self, "_polling_teardown_started", False):
+            return
+        if not self._polling_progress_accepting:
+            return
+        if generation != self._polling_generation:
+            return
+        self._polling_progress_event.set()
+        self._polling_network_error_count = 0
+        self._polling_conflict_count = 0
+        self._send_path_degraded = False
+
+    def _instrument_polling_request(self, request):
+        """Wrap one dedicated PTB getUpdates request with progress tracking."""
+        do_request = request.do_request
+
+        async def _do_request(*args, **kwargs):
+            generation = _POLLING_GENERATION_CONTEXT.get()
+            result = await do_request(*args, **kwargs)
+            status_code, payload = result
+            if generation is not None and 200 <= status_code < 300:
+                try:
+                    # Use the request's own parser so health observation agrees
+                    # exactly with PTB's authoritative response handling (e.g.
+                    # UTF-8 replacement decoding and BOM rejection).
+                    envelope = request.parse_json_payload(payload)
+                except Exception:
+                    # Instrumentation is observational: PTB still parses the
+                    # untouched payload and owns the resulting exception.
+                    pass
+                else:
+                    if (
+                        isinstance(envelope, dict)
+                        and envelope.get("ok") is True
+                        and "result" in envelope
+                    ):
+                        self._record_polling_progress(generation)
+            return result
+
+        request.do_request = _do_request
+        return request
+
+    async def _start_polling_once(
+        self,
+        app,
+        *,
+        drop_pending_updates: bool,
+        error_callback,
+    ) -> None:
+        """Start one generation and verify real getUpdates progress."""
+        if getattr(self, "_polling_teardown_started", False):
+            raise _PollingLifecycleAbort("Telegram polling teardown started")
+        generation, progress = self._begin_polling_generation()
+        if not self._polling_progress_accepting:
+            raise _PollingLifecycleAbort("Telegram polling teardown started")
+
+        def _generation_error_callback(error: Exception) -> None:
+            if getattr(self, "_polling_teardown_started", False):
+                return
+            if generation != self._polling_generation:
+                return
+            if error_callback is not None:
+                callback_context_token = [redacted:secret-assignment]
+                try:
+                    error_callback(error)
+                finally:
+                    _POLLING_GENERATION_CONTEXT.reset(callback_context_token)
+
+        context_token = [redacted:secret-assignment]
+        try:
+            await asyncio.wait_for(
+                app.updater.start_polling(
+                    allowed_updates=Update.ALL_TYPES,
+                    drop_pending_updates=drop_pending_updates,
+                    error_callback=_generation_error_callback,
+                ),
+                timeout=_UPDATER_START_TIMEOUT,
+            )
+        finally:
+            _POLLING_GENERATION_CONTEXT.reset(context_token)
+        if getattr(self, "_polling_teardown_started", False):
+            self._polling_progress_accepting = False
+            self._send_path_degraded = True
+            raise _PollingLifecycleAbort("Telegram polling teardown started")
+        self._schedule_polling_progress_verifier(generation, progress)
+
+    def _schedule_polling_progress_verifier(
+        self, generation: int, progress: asyncio.Event
+    ) -> None:
+        """Own exactly one tracked verifier for the current generation."""
+        if getattr(self, "_polling_teardown_started", False):
+            self._polling_progress_accepting = False
+            self._send_path_degraded = True
+            return
+        previous = getattr(self, "_polling_progress_verifier_task", None)
+        if previous is not None and not previous.done():
+            previous.cancel()
+
+        task = asyncio.get_running_loop().create_task(
+            self._verify_polling_after_reconnect(generation, progress)
+        )
+        self._polling_progress_verifier_task = task
+        self._background_tasks.add(task)
+
+        def _clear_finished_verifier(finished: asyncio.Task) -> None:
+            self._background_tasks.discard(finished)
+            if self._polling_progress_verifier_task is finished:
+                self._polling_progress_verifier_task = None
+
+        task.add_done_callback(_clear_finished_verifier)
+
     def _get_general_request_drain_lock(self) -> asyncio.Lock:
         lock = getattr(self, "_general_request_drain_lock", None)
         if lock is None:
@@ -1996,6 +2144,8 @@ def _schedule_polling_recovery(self, error: Exception, *, reason: str) -> None:
         adapter: the gateway process stays alive and the existing reconnect
         ladder (``_handle_polling_network_error``) recovers in the background.
         """
+        if getattr(self, "_polling_teardown_started", False):
+            return
         if self.has_fatal_error:
             return
         if self._polling_error_task and not self._polling_error_task.done():
@@ -2047,6 +2197,8 @@ async def _start_polling_resilient(self, *, drop_pending_updates: bool, error_ca
         network error was scheduled for background recovery instead of raising
         (keeping the gateway process alive).
         """
+        if getattr(self, "_polling_teardown_started", False):
+            return False
         if not (self._app and self._app.updater):
             raise RuntimeError("Telegram application/updater not initialized")
         try:
@@ -2056,16 +2208,17 @@ async def _start_polling_resilient(self, *, drop_pending_updates: bool, error_ca
             # TimeoutError (OSError subclass), so the except below classifies
             # it via _looks_like_network_error and schedules background
             # recovery instead of blocking connect() indefinitely.
-            await asyncio.wait_for(
-                self._app.updater.start_polling(
-                    allowed_updates=Update.ALL_TYPES,
-                    drop_pending_updates=drop_pending_updates,
-                    error_callback=error_callback,
-                ),
-                timeout=_UPDATER_START_TIMEOUT,
+            await self._start_polling_once(
+                self._app,
+                drop_pending_updates=drop_pending_updates,
+                error_callback=error_callback,
             )
             return True
+        except _PollingLifecycleAbort:
+            return False
         except Exception as err:
+            if getattr(self, "_polling_teardown_started", False):
+                return False
             if self._looks_like_polling_conflict(err):
                 logger.warning(
                     "[%s] Telegram polling bootstrap conflict; gateway stays alive "
@@ -2094,6 +2247,8 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
         MAX_NETWORK_RETRIES attempts, then mark the adapter retryable-fatal so
         the supervisor restarts the gateway process.
         """
+        if getattr(self, "_polling_teardown_started", False):
+            return
         if self.has_fatal_error:
             return
 
@@ -2123,6 +2278,9 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
         )
         await asyncio.sleep(delay)
 
+        if getattr(self, "_polling_teardown_started", False):
+            return
+
         # Capture a stable local reference: self._app can be reassigned to None
         # by a concurrent disconnect() while we're suspended across the awaits
         # below, and re-reading self._app after that point would silently swap
@@ -2153,64 +2311,39 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
         except Exception:
             pass
 
+        if getattr(self, "_polling_teardown_started", False):
+            return
         await self._drain_polling_connections()
 
+        if getattr(self, "_polling_teardown_started", False):
+            return
+
         try:
             if not app:
                 raise RuntimeError("Telegram application was torn down during reconnect")
-            # Guard start_polling() with a timeout: when the connection pool is
-            # in a degraded state (e.g., after _drain_polling_connections()), the
-            # httpx client may hold a stale socket that neither connects nor times
-            # out within PTB's internal flow. Bounding start_polling() prevents
-            # the reconnect ladder from stalling indefinitely and allows the
-            # heartbeat loop to trigger its own recovery path.
-            # Refs: NousResearch/hermes-agent#59614
-            try:
-                await asyncio.wait_for(
-                    app.updater.start_polling(
-                        allowed_updates=Update.ALL_TYPES,
-                        drop_pending_updates=False,
-                        error_callback=self._polling_error_callback_ref,
-                    ),
-                    timeout=_UPDATER_START_TIMEOUT,
-                )
-            except asyncio.TimeoutError:
-                raise RuntimeError(
-                    "start_polling() timed out — connection pool may be wedged"
-                )
+            await self._start_polling_once(
+                app,
+                drop_pending_updates=False,
+                error_callback=self._polling_error_callback_ref,
+            )
             logger.info(
-                "[%s] Telegram polling resumed after network error (attempt %d)",
+                "[%s] Telegram polling restarted after network error (attempt %d); "
+                "health pending getUpdates progress",
                 self.name, attempt,
             )
-            self._polling_network_error_count = 0
-            # start_polling() succeeding IS the recovery signal: the long-poll
-            # connection is live again, so clear the degraded flag immediately
-            # rather than blocking all outbound sends for the full
-            # HEARTBEAT_PROBE_DELAY window. The deferred probe below is a
-            # defensive re-check — if it later detects a silent wedge (PTB
-            # running=True but consumer task dead) it re-enters the ladder,
-            # which re-sets _send_path_degraded. Without this clear here, a
-            # clean reconnect leaves the flag stuck True until the 60s probe
-            # (or forever, if the probe is never scheduled), blocking the send
-            # path even though the bot has fully recovered. See #35205.
-            self._send_path_degraded = False
-            # start_polling() returning is necessary but not sufficient:
-            # PTB's Updater can be left in a state where `running` is True
-            # but the underlying long-poll task is wedged on a stale httpx
-            # connection and never makes progress. No error_callback fires
-            # in that state, so the reconnect ladder won't advance on its
-            # own. Schedule a deferred probe to detect the wedge and
-            # re-enter the ladder if needed.
-            if not self.has_fatal_error:
-                probe = asyncio.ensure_future(self._verify_polling_after_reconnect())
-                self._background_tasks.add(probe)
-                probe.add_done_callback(self._background_tasks.discard)
+        except _PollingLifecycleAbort:
+            return
         except Exception as retry_err:
+            if getattr(self, "_polling_teardown_started", False):
+                return
             safe_retry_error = _redact_telegram_error_text(retry_err)
             logger.warning("[%s] Telegram polling reconnect failed: %s", self.name, safe_retry_error)
             # start_polling failed — polling is dead and no further error
             # callbacks will fire, so schedule the next retry ourselves.
-            if not self.has_fatal_error:
+            if (
+                not self.has_fatal_error
+                and not getattr(self, "_polling_teardown_started", False)
+            ):
                 task = asyncio.ensure_future(
                     self._handle_polling_network_error(retry_err)
                 )
@@ -2240,9 +2373,9 @@ async def _polling_heartbeat_loop(self) -> None:
         ``_handle_polling_network_error`` — the same path triggered by PTB's own
         ``error_callback`` — which drains the dead pool and restarts polling.
 
-        Unlike ``_verify_polling_after_reconnect`` (a one-shot probe scheduled
-        only after an explicit reconnect), this loop runs for the full lifetime
-        of the polling connection, so it catches a socket that wedges during
+        Unlike the generation verifier (a one-shot progress deadline after
+        every polling start), this loop runs for the full lifetime of the
+        polling connection, so it catches a socket that wedges later during
         steady-state operation without any prior error event.
         """
         HEARTBEAT_INTERVAL = 90   # seconds between probes
@@ -2251,6 +2384,8 @@ async def _polling_heartbeat_loop(self) -> None:
         while True:
             try:
                 await asyncio.sleep(HEARTBEAT_INTERVAL)
+                if getattr(self, "_polling_teardown_started", False):
+                    return
                 if self.has_fatal_error:
                     return
                 bot = self._app.bot if self._app else None
@@ -2307,6 +2442,8 @@ async def _probe_pending_updates(self, bot, probe_timeout: float) -> None:
         report a queue against a live consumer. We detect the stopped updater
         directly and feed the same ladder (#55769).
         """
+        if getattr(self, "_polling_teardown_started", False):
+            return
         # Only meaningful in polling mode; in webhook mode Telegram pushes
         # updates and holds no server-side queue.
         if self._webhook_mode:
@@ -2340,6 +2477,8 @@ async def _probe_pending_updates(self, bot, probe_timeout: float) -> None:
             )
             if self._polling_not_running_count >= 2:
                 self._polling_not_running_count = 0
+                if getattr(self, "_polling_teardown_started", False):
+                    return
                 logger.warning(
                     "[%s] Telegram updater is not running (long-poll task "
                     "gone); triggering polling restart",
@@ -2374,6 +2513,8 @@ async def _probe_pending_updates(self, bot, probe_timeout: float) -> None:
         )
         if self._polling_pending_stuck_count >= 2:
             self._polling_pending_stuck_count = 0
+            if getattr(self, "_polling_teardown_started", False):
+                return
             logger.warning(
                 "[%s] getUpdates consumer appears wedged (queue not draining); "
                 "triggering polling restart",
@@ -2386,66 +2527,96 @@ async def _probe_pending_updates(self, bot, probe_timeout: float) -> None:
                 )
             )
 
-    async def _verify_polling_after_reconnect(self) -> None:
-        """Heartbeat probe scheduled after a successful reconnect.
-
-        PTB's Updater can survive a botched stop()+start_polling() cycle
-        with `running=True` but a wedged consumer task. No error callback
-        fires, so the reconnect ladder doesn't advance on its own. This
-        probe detects the wedge by:
-
-        1. Sleeping HEARTBEAT_PROBE_DELAY so a healthy long-poll has time
-           to complete at least one cycle.
-        2. Verifying `Updater.running` is still True.
-        3. Probing the bot endpoint with a tight asyncio timeout. A
-           wedged httpx pool fails this probe; a healthy one returns
-           well under the timeout.
-
-        On connectivity failure, re-enter the reconnect ladder (via
-        ``_schedule_polling_recovery`` so the in-flight guard and task
-        bookkeeping apply) and let the existing MAX_NETWORK_RETRIES path
-        ultimately escalate to fatal-error. Auth/validation failures
-        (``InvalidToken``, ``BadRequest``, ...) are not connectivity symptoms
-        and must not trigger reconnect churn — same policy as the heartbeat
-        loop and the pending-update probe (#62098, #63243).
+    async def _verify_polling_after_reconnect(
+        self,
+        generation: Optional[int] = None,
+        progress: Optional[asyncio.Event] = None,
+    ) -> None:
+        """Require getUpdates progress, using getMe only to classify failure.
+
+        The generation-bound event is set only by a successful response on the
+        dedicated getUpdates request. A general-path getMe success can classify
+        connectivity, but cannot heal polling health. Connectivity failures
+        enter the guarded recovery ladder; auth/validation errors do not churn.
         """
-        HEARTBEAT_PROBE_DELAY = 60
         PROBE_TIMEOUT = 10
+        if getattr(self, "_polling_teardown_started", False):
+            return
+        if generation is None:
+            generation = self._polling_generation
+        if progress is None:
+            progress = self._polling_progress_event
 
-        await asyncio.sleep(HEARTBEAT_PROBE_DELAY)
+        try:
+            await asyncio.wait_for(
+                progress.wait(), timeout=_POLLING_PROGRESS_TIMEOUT
+            )
+        except asyncio.TimeoutError:
+            pass
 
-        if self.has_fatal_error:
+        if getattr(self, "_polling_teardown_started", False):
+            return
+        if progress.is_set() or self.has_fatal_error:
+            return
+        if not self._polling_progress_accepting:
+            return
+        if generation != self._polling_generation:
             return
-        if not (self._app and self._app.updater and self._app.updater.running):
+        if progress is not self._polling_progress_event:
+            return
+
+        app = self._app
+        if not (app and app.updater and app.updater.running):
             logger.warning(
-                "[%s] Updater not running %ds after reconnect — treating as wedged",
-                self.name, HEARTBEAT_PROBE_DELAY,
+                "[%s] Updater made no getUpdates progress and is not running",
+                self.name,
             )
             self._schedule_polling_recovery(
-                RuntimeError("Updater not running after reconnect heartbeat"),
-                reason="post-reconnect probe: updater not running",
+                RuntimeError("Updater not running after polling progress deadline"),
+                reason="polling progress verifier: updater not running",
             )
             return
 
         try:
-            await asyncio.wait_for(self._app.bot.get_me(), PROBE_TIMEOUT)
-            self._send_path_degraded = False
+            await asyncio.wait_for(app.bot.get_me(), PROBE_TIMEOUT)
         except Exception as probe_err:
+            if getattr(self, "_polling_teardown_started", False):
+                return
+            if self.has_fatal_error or not self._polling_progress_accepting:
+                return
+            if generation != self._polling_generation:
+                return
+            if progress is not self._polling_progress_event or progress.is_set():
+                return
             if not self._looks_like_network_error(probe_err):
                 logger.warning(
-                    "[%s] Post-reconnect probe hit a non-connectivity error"
+                    "[%s] Polling progress verifier hit a non-connectivity error"
                     " (not retrying): %s",
                     self.name, _redact_telegram_error_text(probe_err),
                 )
                 return
             logger.warning(
-                "[%s] Polling heartbeat probe failed %ds after reconnect: %s",
-                self.name, HEARTBEAT_PROBE_DELAY,
-                _redact_telegram_error_text(probe_err),
+                "[%s] Polling progress verifier connectivity probe failed: %s",
+                self.name, _redact_telegram_error_text(probe_err),
             )
             self._schedule_polling_recovery(
-                probe_err, reason="post-reconnect probe failure"
+                probe_err,
+                reason="polling progress verifier connectivity failure",
             )
+            return
+
+        if getattr(self, "_polling_teardown_started", False):
+            return
+        if self.has_fatal_error or not self._polling_progress_accepting:
+            return
+        if generation != self._polling_generation:
+            return
+        if progress is not self._polling_progress_event or progress.is_set():
+            return
+        self._schedule_polling_recovery(
+            RuntimeError("getUpdates made no progress before verifier deadline"),
+            reason="polling progress verifier: general path healthy but getUpdates stalled",
+        )
 
     def _disarm_ptb_retry_loop(self) -> None:
         """Synchronously stop PTB's internal polling retry loop.
@@ -2511,6 +2682,8 @@ def _disarm_ptb_retry_loop(self) -> None:
         )
 
     async def _handle_polling_conflict(self, error: Exception) -> None:
+        if getattr(self, "_polling_teardown_started", False):
+            return
         if self.has_fatal_error and self.fatal_error_code == "telegram_polling_conflict":
             return
         # Transient 409 Conflict errors arise when the previous gateway process
@@ -2566,7 +2739,11 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 pass
 
             await asyncio.sleep(RETRY_DELAY)
+            if getattr(self, "_polling_teardown_started", False):
+                return
             await self._drain_polling_connections()
+            if getattr(self, "_polling_teardown_started", False):
+                return
 
             # Capture a stable local reference: self._app can be reassigned to
             # None by a concurrent disconnect() while we're suspended across
@@ -2578,31 +2755,22 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
             try:
                 if not app:
                     raise RuntimeError("Telegram application was torn down during conflict reconnect")
-                # Same watchdog bound as the network-error ladder: an
-                # exhausted pool hangs start_polling() on the conflict path
-                # identically (#59614). Timeout converts to RuntimeError so
-                # the except below logs a readable message and schedules the
-                # next conflict attempt instead of wedging attempt N forever.
-                try:
-                    await asyncio.wait_for(
-                        app.updater.start_polling(
-                            allowed_updates=Update.ALL_TYPES,
-                            drop_pending_updates=False,
-                            error_callback=self._polling_error_callback_ref,
-                        ),
-                        timeout=_UPDATER_START_TIMEOUT,
-                    )
-                except asyncio.TimeoutError:
-                    raise RuntimeError(
-                        "start_polling() timed out — connection pool may be wedged"
-                    )
+                await self._start_polling_once(
+                    app,
+                    drop_pending_updates=False,
+                    error_callback=self._polling_error_callback_ref,
+                )
                 logger.info(
-                    "[%s] Telegram polling resumed after conflict retry %d/%d",
+                    "[%s] Telegram polling restarted after conflict retry %d/%d; "
+                    "health pending getUpdates progress",
                     self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES,
                 )
-                self._polling_conflict_count = 0  # reset counter on success
+                return
+            except _PollingLifecycleAbort:
                 return
             except Exception as retry_err:
+                if getattr(self, "_polling_teardown_started", False):
+                    return
                 logger.warning(
                     "[%s] Telegram polling retry %d/%d failed: %s. "
                     "Scheduling next attempt.",
@@ -2614,7 +2782,10 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                 # a fatal error leaves the adapter in a limbo state: the
                 # gateway process is alive and reports "connected" but
                 # no messages are received or sent.
-                if self._polling_conflict_count < MAX_CONFLICT_RETRIES:
+                if (
+                    self._polling_conflict_count < MAX_CONFLICT_RETRIES
+                    and not getattr(self, "_polling_teardown_started", False)
+                ):
                     # We are inside a running coroutine, so the running loop is
                     # guaranteed to exist. asyncio.get_event_loop() is deprecated
                     # and raises "RuntimeError: There is no current event loop in
@@ -2628,6 +2799,9 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                     return
                 # Fall through to fatal on the last retry.
 
+        if getattr(self, "_polling_teardown_started", False):
+            return
+
         # Exhausted all retries — declare a fatal error so the gateway
         # runner can surface this clearly and the user knows to act.
         message = (
@@ -3084,6 +3258,14 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             TELEGRAM_WEBHOOK_PORT   Local listen port (default 8443)
             TELEGRAM_WEBHOOK_SECRET Secret token for update verification
         """
+        # Explicit connect() is the only operation allowed to reopen polling
+        # after a completed, serialized teardown. Background recovery never
+        # clears this fence.
+        self._polling_teardown_started = False
+        # Mode selection is re-evaluated on every explicit connection. Keep
+        # webhook state false unless this connection starts its webhook.
+        self._webhook_mode = False
+
         if not TELEGRAM_AVAILABLE:
             logger.error(
                 "[%s] python-telegram-bot not installed. Run: pip install python-telegram-bot",
@@ -3246,6 +3428,7 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
                     **request_kwargs, httpx_kwargs=_with_limits()
                 )
 
+            get_updates_request = self._instrument_polling_request(get_updates_request)
             builder = builder.request(request).get_updates_request(get_updates_request)
             self._app = builder.build()
             self._bot = self._app.bot
@@ -3378,6 +3561,8 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
                     drop_pending_updates=not is_reconnect,
                 )
                 self._webhook_mode = True
+                self._polling_progress_accepting = False
+                self._send_path_degraded = False
                 logger.info(
                     "[%s] Webhook server listening on 0.0.0.0:%d%s",
                     self.name, webhook_port, webhook_path,
@@ -3394,6 +3579,8 @@ def _with_limits(httpx_kwargs: Optional[dict] = None) -> dict:
                 loop = asyncio.get_running_loop()
 
                 def _polling_error_callback(error: Exception) -> None:
+                    if getattr(self, "_polling_teardown_started", False):
+                        return
                     if self._polling_error_task and not self._polling_error_task.done():
                         return
                     if self._looks_like_polling_conflict(error):
@@ -3527,7 +3714,8 @@ def collect(task: Optional[asyncio.Task]) -> None:
             collect(task)
         for task in list(self._pending_text_batch_tasks.values()):
             collect(task)
-        collect(self._polling_error_task)
+        collect(getattr(self, "_polling_error_task", None))
+        collect(getattr(self, "_polling_progress_verifier_task", None))
 
         for task in pending_tasks:
             task.cancel()
@@ -3540,8 +3728,10 @@ def collect(task: Optional[asyncio.Task]) -> None:
         self._pending_photo_batches.clear()
         self._pending_text_batch_tasks.clear()
         self._pending_text_batches.clear()
-        if self._polling_error_task is not current_task:
+        if getattr(self, "_polling_error_task", None) is not current_task:
             self._polling_error_task = None
+        if getattr(self, "_polling_progress_verifier_task", None) is not current_task:
+            self._polling_progress_verifier_task = None
 
     async def disconnect(self) -> None:
         """Stop polling/webhook, cancel pending delayed deliveries, and disconnect."""
@@ -3549,6 +3739,42 @@ async def disconnect(self) -> None:
         # that wins the race against teardown and prevents new delayed tasks
         # from being scheduled by late update handlers.
         self._mark_disconnected()
+        self._polling_teardown_started = True
+        self._polling_progress_accepting = False
+        self._polling_generation = getattr(self, "_polling_generation", 0) + 1
+        self._polling_progress_event = asyncio.Event()
+        self._send_path_degraded = True
+
+        # Recovery can be suspended in stop/drain/start while disconnect begins.
+        # Cancel and await both polling lifecycle owners immediately after the
+        # fence, before any other teardown await lets them start a new generation.
+        current_task = asyncio.current_task()
+        lifecycle_tasks: list[asyncio.Task] = []
+        lifecycle_seen: set[int] = set()
+        for task in (
+            getattr(self, "_polling_error_task", None),
+            getattr(self, "_polling_progress_verifier_task", None),
+        ):
+            if not task or task.done() or task is current_task:
+                continue
+            marker = id(task)
+            if marker in lifecycle_seen:
+                continue
+            lifecycle_seen.add(marker)
+            task.cancel()
+            if asyncio.isfuture(task) or asyncio.iscoroutine(task):
+                lifecycle_tasks.append(task)
+        if lifecycle_tasks:
+            await asyncio.gather(*lifecycle_tasks, return_exceptions=True)
+        if getattr(self, "_polling_error_task", None) is not current_task:
+            self._polling_error_task = None
+        if getattr(self, "_polling_progress_verifier_task", None) is not current_task:
+            self._polling_progress_verifier_task = None
+
+        # Cancellation callbacks may have run while awaited; the teardown fence
+        # remains authoritative regardless of their finalizers.
+        self._polling_progress_accepting = False
+        self._send_path_degraded = True
 
         # Cancel deferred post-connect housekeeping (command-menu / DM-topic /
         # status-indicator Bot API calls) so it cannot fire into a half-torn-down
@@ -3562,10 +3788,11 @@ async def disconnect(self) -> None:
 
         # Cancel the heartbeat before tearing down the app so the probe task
         # cannot fire get_me() into a half-shutdown bot client.
-        if self._polling_heartbeat_task and not self._polling_heartbeat_task.done():
-            self._polling_heartbeat_task.cancel()
+        polling_heartbeat_task = getattr(self, "_polling_heartbeat_task", None)
+        if polling_heartbeat_task and not polling_heartbeat_task.done():
+            polling_heartbeat_task.cancel()
             try:
-                await self._polling_heartbeat_task
+                await polling_heartbeat_task
             except asyncio.CancelledError:
                 pass
         self._polling_heartbeat_task = None

scripts/release.py

@@ -79,6 +79,7 @@
     "[redacted:email]": "Umi4Life",  # PR #47377 salvage (agent: emit one-shot fallback switch notice on successful fallback so gateway users see model/provider change; #35419)
     "[redacted:email]": "austinlaw076",  # PR #57563 partial salvage (auth: lazy per-profile Anthropic OAuth file; gateway: whatsapp_cloud/line added to port-binding platform set)
     "[redacted:email]": "liuhao1024",  # PR #56993 salvage (gateway: process-level HERMES_HOME for pid/lock/status identity files; #56986)
+    "[redacted:email]": "Roseyco-management",  # PR #63581 salvage (telegram: require getUpdates progress before polling is healthy; #63243, #63766)
     "[redacted:email]": "gauravsaxena1997",  # PR #59868 partial salvage (agent: guard response.text against httpx.ResponseNotRead in _summarize_api_error; #59769)
     "[redacted:email]": "irresi",  # PR #59048 salvage (gateway: scope reset banners' session info to the serving profile; #59003)
     "[redacted:email]": "s905060",  # PR #57943 salvage (photon: auto-reinstall stale sidecar node_modules when lockfile is newer than npm's install marker; #59169)

tests/gateway/test_telegram_conflict.py

@@ -137,14 +137,13 @@ async def fake_start_polling(**kwargs):
 
     # First conflict: should retry, NOT be fatal
     captured["error_callback"](conflict("Conflict: terminated by other getUpdates request"))
-    await asyncio.sleep(0)
-    await asyncio.sleep(0)
-    # Give the scheduled task a chance to run
-    for _ in range(10):
-        await asyncio.sleep(0)
+    await adapter._polling_error_task
 
     assert adapter.has_fatal_error is False, "First conflict should not be fatal"
-    assert adapter._polling_conflict_count == 0, "Count should reset after successful retry"
+    assert adapter._polling_conflict_count == 1, (
+        "Count must remain until the retried generation makes getUpdates progress"
+    )
+    assert adapter._send_path_degraded is True
 
     # connect() now starts a lifetime _polling_heartbeat_loop task. With
     # asyncio.sleep mocked to instant above, it must not be left running or it
@@ -152,6 +151,54 @@ async def fake_start_polling(**kwargs):
     await _cancel_heartbeat(adapter)
 
 
[redacted:email]
+async def test_current_generation_conflicts_accumulate_after_start_returns(monkeypatch):
+    """A later async 409 must advance the retry ladder after PTB start returns."""
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
+    callbacks = []
+    conflict_tasks = []
+
+    async def capture_start(**kwargs):
+        callbacks.append(kwargs["error_callback"])
+
+    updater = SimpleNamespace(
+        start_polling=AsyncMock(side_effect=capture_start),
+        stop=AsyncMock(),
+        running=False,
+    )
+    app = SimpleNamespace(updater=updater)
+    adapter._app = app
+    adapter._drain_polling_connections = AsyncMock()
+    monkeypatch.setattr("asyncio.sleep", AsyncMock())
+
+    def dispatch_conflict(error):
+        conflict_tasks.append(
+            asyncio.create_task(adapter._handle_polling_conflict(error))
+        )
+
+    adapter._polling_error_callback_ref = dispatch_conflict
+    await adapter._start_polling_once(
+        app,
+        drop_pending_updates=False,
+        error_callback=dispatch_conflict,
+    )
+    conflict = type("Conflict", (Exception,), {})
+
+    try:
+        callbacks[0](conflict("first async conflict"))
+        await conflict_tasks[-1]
+        assert adapter._polling_conflict_count == 1
+
+        callbacks[1](conflict("second async conflict"))
+        await conflict_tasks[-1]
+        assert adapter._polling_conflict_count == 2
+    finally:
+        verifier = adapter._polling_progress_verifier_task
+        if verifier is not None and not verifier.done():
+            verifier.cancel()
+            await asyncio.gather(verifier, return_exceptions=True)
+
+
 @pytest.mark.asyncio
 async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
     """After exhausting retries, the conflict should become fatal."""

tests/gateway/test_telegram_network_reconnect.py

@@ -35,6 +35,7 @@ def _ensure_telegram_mock():
 
 _ensure_telegram_mock()
 
+from plugins.platforms.telegram import adapter as tg_adapter  # noqa: E402
 from plugins.platforms.telegram.adapter import TelegramAdapter  # noqa: E402
 
 
@@ -50,6 +51,13 @@ def _make_adapter() -> TelegramAdapter:
     return TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
 
 
+async def _complete_current_polling_generation(adapter: TelegramAdapter) -> None:
+    verifier = adapter._polling_progress_verifier_task
+    adapter._record_polling_progress(adapter._polling_generation)
+    if verifier is not None:
+        await verifier
+
+
 @pytest.mark.asyncio
 async def test_reconnect_self_schedules_on_start_polling_failure():
     """
@@ -157,9 +165,9 @@ async def test_reconnect_chained_retry_updates_polling_error_task():
 
 
 @pytest.mark.asyncio
-async def test_reconnect_success_resets_error_count():
+async def test_reconnect_success_waits_for_progress_to_reset_error_count():
     """
-    When start_polling() succeeds, _polling_network_error_count should reset to 0.
+    start_polling() return alone cannot reset the network-error count.
     """
     adapter = _make_adapter()
     adapter._polling_network_error_count = 3
@@ -177,7 +185,12 @@ async def test_reconnect_success_resets_error_count():
     with patch("asyncio.sleep", new_callable=AsyncMock):
         await adapter._handle_polling_network_error(Exception("Bad Gateway"))
 
+    assert adapter._polling_network_error_count == 4
+    assert adapter._send_path_degraded is True
+
+    await _complete_current_polling_generation(adapter)
     assert adapter._polling_network_error_count == 0
+    assert adapter._send_path_degraded is False
 
     # Clean up the heartbeat-probe task scheduled after a successful reconnect.
     pending = [t for t in adapter._background_tasks if not t.done()]
@@ -263,6 +276,8 @@ async def test_reconnect_drains_polling_request_only():
 
     # Reconnect must still succeed
     mock_app.updater.start_polling.assert_called_once()
+    assert adapter._polling_network_error_count == 2
+    await _complete_current_polling_generation(adapter)
     assert adapter._polling_network_error_count == 0
 
 
@@ -283,6 +298,8 @@ async def test_reconnect_continues_if_drain_fails():
 
     # start_polling must still be called despite drain failure
     mock_app.updater.start_polling.assert_called_once()
+    assert adapter._polling_network_error_count == 2
+    await _complete_current_polling_generation(adapter)
     assert adapter._polling_network_error_count == 0
 
 
@@ -339,10 +356,9 @@ async def test_drain_helper_noop_without_app():
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_no_op_when_polling_healthy():
+async def test_polling_verifier_exits_on_matching_progress(monkeypatch):
     """
-    Probe scheduled after a successful reconnect: Updater.running=True and
-    bot.get_me() returns quickly → recovery confirmed, no further action.
+    Matching getUpdates progress exits without probing the general path.
     """
     adapter = _make_adapter()
 
@@ -355,19 +371,20 @@ async def test_heartbeat_probe_no_op_when_polling_healthy():
     adapter._app = mock_app
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    adapter._record_polling_progress(generation)
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
-    mock_app.bot.get_me.assert_awaited_once()
+    mock_app.bot.get_me.assert_not_awaited()
     adapter._handle_polling_network_error.assert_not_awaited()
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_reenters_ladder_when_updater_not_running():
+async def test_heartbeat_probe_reenters_ladder_when_updater_not_running(monkeypatch):
     """
-    If Updater.running has flipped to False by the heartbeat delay, treat
-    as wedged: re-enter the reconnect ladder.
+    If Updater.running is False at the progress deadline, re-enter recovery.
     """
     adapter = _make_adapter()
 
@@ -380,9 +397,10 @@ async def test_heartbeat_probe_reenters_ladder_when_updater_not_running():
     adapter._app = mock_app
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
     mock_app.bot.get_me.assert_not_called()
     # Recovery is scheduled through _schedule_polling_recovery (#63243), so
@@ -397,7 +415,7 @@ async def test_heartbeat_probe_reenters_ladder_when_updater_not_running():
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out():
+async def test_heartbeat_probe_reenters_ladder_when_get_me_times_out(monkeypatch):
     """
     If bot.get_me() hangs longer than PROBE_TIMEOUT, treat as wedged.
     Simulates the connection-pool wedge that motivated this fix.
@@ -416,15 +434,16 @@ async def hang_forever(*args, **kwargs):
     adapter._app = mock_app
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
     async def fast_wait_for(coro, timeout):
         if asyncio.iscoroutine(coro):
             coro.close()
         raise asyncio.TimeoutError()
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", new=fast_wait_for):
-            await adapter._verify_polling_after_reconnect()
+    with patch("plugins.platforms.telegram.adapter.asyncio.wait_for", new=fast_wait_for):
+        await adapter._verify_polling_after_reconnect(generation, progress)
 
     task = adapter._polling_error_task
     assert task is not None
@@ -433,7 +452,7 @@ async def fast_wait_for(coro, timeout):
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error():
+async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error(monkeypatch):
     """
     Any exception raised by bot.get_me() (NetworkError, ConnectionError, etc.)
     should re-enter the reconnect ladder with the original exception.
@@ -449,9 +468,10 @@ async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error():
     adapter._app = mock_app
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
     task = adapter._polling_error_task
     assert task is not None
@@ -466,7 +486,7 @@ async def test_heartbeat_probe_reenters_ladder_on_get_me_network_error():
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_ignores_auth_errors():
+async def test_heartbeat_probe_ignores_auth_errors(monkeypatch):
     """
     Auth/validation failures from the post-reconnect probe must not enter the
     network-reconnect ladder (#63243): a revoked token would otherwise churn
@@ -487,16 +507,17 @@ async def test_heartbeat_probe_ignores_auth_errors():
     adapter._app = mock_app
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
     assert adapter._polling_error_task is None
     adapter._handle_polling_network_error.assert_not_awaited()
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_defers_to_inflight_recovery():
+async def test_heartbeat_probe_defers_to_inflight_recovery(monkeypatch):
     """
     A probe failure while another recovery is mid-flight must not start a
     second concurrent stop/drain/start_polling sequence (#63243) — overlapping
@@ -517,16 +538,17 @@ async def test_heartbeat_probe_defers_to_inflight_recovery():
     adapter._polling_error_task = inflight
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
     assert adapter._polling_error_task is inflight
     adapter._handle_polling_network_error.assert_not_awaited()
 
 
 @pytest.mark.asyncio
-async def test_heartbeat_probe_skips_when_already_fatal():
+async def test_heartbeat_probe_skips_when_already_fatal(monkeypatch):
     """
     If the adapter is already in fatal-error state by the time the probe
     delay elapses, the probe should bail without further action.
@@ -539,9 +561,10 @@ async def test_heartbeat_probe_skips_when_already_fatal():
     adapter._app = mock_app
 
     adapter._handle_polling_network_error = AsyncMock()
+    generation, progress = adapter._begin_polling_generation()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0)
 
-    with patch("asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
     mock_app.bot.get_me.assert_not_called()
     adapter._handle_polling_network_error.assert_not_awaited()
@@ -1060,4 +1083,3 @@ async def _fake_start_polling(**kwargs):
     # The reconnect ladder must have advanced past the hung stop().
     assert drain_called, "_drain_polling_connections was not called after stop() timeout"
     assert start_polling_called, "start_polling was not called after stop() timeout"
-

tests/gateway/test_telegram_polling_progress.py

@@ -0,0 +1,774 @@
+"""Behavior contract for generation-safe Telegram polling progress."""
+
+import asyncio
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+from plugins.platforms.telegram import adapter as tg_adapter
+from plugins.platforms.telegram.adapter import TelegramAdapter
+
+
+class _ControlledRequest:
+    """Minimal PTB request double with controllable completion."""
+
+    instances = []
+
+    @staticmethod
+    def parse_json_payload(payload):
+        """Match PTB's response authority used by the progress observer."""
+        return json.loads(payload.decode("utf-8", "replace"))
+
+    def __init__(self, *args, result=None, error=None, entered=None, release=None, **kwargs):
+        self.result = result
+        self.error = error
+        self.entered = entered
+        self.release = release
+        self.args = args
+        self.kwargs = kwargs
+        type(self).instances.append(self)
+
+    async def do_request(self, *args, **kwargs):
+        if self.entered is not None:
+            self.entered.set()
+        if self.release is not None:
+            await self.release.wait()
+        if self.error is not None:
+            raise self.error
+        return self.result
+
+
+def _make_adapter() -> TelegramAdapter:
+    return TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+
+
+def _mock_polling_app(*, get_me=None):
+    app = MagicMock()
+    app.updater = MagicMock()
+    app.updater.running = True
+    app.updater.stop = AsyncMock()
+    app.updater.start_polling = AsyncMock()
+    app.bot = MagicMock()
+    app.bot.get_me = get_me or AsyncMock(return_value=MagicMock())
+    app.running = False
+    app.shutdown = AsyncMock()
+    return app
+
+
+class _LifecycleBuilder:
+    def __init__(self, app):
+        self.app = app
+        self.polling_request = None
+
+    def token(self, _token):
+        return self
+
+    def request(self, _request):
+        return self
+
+    def get_updates_request(self, request):
+        self.polling_request = request
+        return self
+
+    def build(self):
+        return self.app
+
+
+def _lifecycle_app():
+    app = MagicMock()
+    app.updater = MagicMock()
+    app.updater.running = True
+    app.updater.start_polling = AsyncMock()
+    app.updater.start_webhook = AsyncMock()
+    app.updater.stop = AsyncMock()
+    app.bot = MagicMock()
+    app.bot.delete_webhook = AsyncMock()
+    app.initialize = AsyncMock()
+    app.start = AsyncMock()
+    app.stop = AsyncMock()
+    app.shutdown = AsyncMock()
+    app.running = True
+    return app
+
+
+def _configure_lifecycle_connect(monkeypatch, adapter, apps):
+    builders = [_LifecycleBuilder(app) for app in apps]
+    remaining = iter(builders)
+
+    class _Application:
+        @staticmethod
+        def builder():
+            return next(remaining)
+
+    async def _no_fallback_ips():
+        return []
+
+    monkeypatch.setattr(tg_adapter, "Application", _Application)
+    monkeypatch.setattr(tg_adapter, "HTTPXRequest", _ControlledRequest)
+    monkeypatch.setattr(tg_adapter, "discover_fallback_ips", _no_fallback_ips)
+    monkeypatch.setattr(tg_adapter, "resolve_proxy_url", lambda *args, **kwargs: None)
+    monkeypatch.setattr(adapter, "_acquire_platform_lock", lambda *args, **kwargs: True)
+    monkeypatch.setattr(adapter, "_release_platform_lock", MagicMock())
+    monkeypatch.setattr(adapter, "_fallback_ips", lambda: [])
+    monkeypatch.setattr(adapter, "_start_post_connect_housekeeping", MagicMock())
+    return builders
+
+
+async def _cancel_task(task):
+    if task is None or task.done():
+        return
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+
+
+async def _request_for_generation(generation, request, *args):
+    """Run a direct request double under the production polling context."""
+    generation_context = tg_adapter._POLLING_GENERATION_CONTEXT
+    token = [redacted:secret-assignment]
+    try:
+        return await request.do_request(*args)
+    finally:
+        generation_context.reset(token)
+
+
[redacted:email]
+async def test_polling_disconnect_webhook_reconnect_heals_webhook_send_path(monkeypatch):
+    adapter = _make_adapter()
+    polling_app = _lifecycle_app()
+    webhook_app = _lifecycle_app()
+    _configure_lifecycle_connect(monkeypatch, adapter, [polling_app, webhook_app])
+    monkeypatch.delenv("TELEGRAM_WEBHOOK_URL", raising=False)
+    monkeypatch.delenv("TELEGRAM_WEBHOOK_SECRET", raising=False)
+
+    assert await adapter.connect() is True
+    assert adapter._webhook_mode is False
+    assert adapter._send_path_degraded is True
+    await adapter.disconnect()
+
+    monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", "https://example.test/telegram")
+    monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", "test-secret")
+    try:
+        assert await adapter.connect(is_reconnect=True) is True
+        webhook_app.updater.start_webhook.assert_awaited_once()
+        assert adapter._webhook_mode is True
+        assert adapter._polling_progress_accepting is False
+        assert adapter._send_path_degraded is False
+    finally:
+        await adapter.disconnect()
+
+
[redacted:email]
+async def test_webhook_disconnect_polling_reconnect_resets_mode_and_waits_for_progress(
+    monkeypatch,
+):
+    adapter = _make_adapter()
+    webhook_app = _lifecycle_app()
+    polling_app = _lifecycle_app()
+    builders = _configure_lifecycle_connect(
+        monkeypatch, adapter, [webhook_app, polling_app]
+    )
+    heartbeat_started = asyncio.Event()
+    heartbeat_modes = []
+
+    async def heartbeat():
+        heartbeat_modes.append(adapter._webhook_mode)
+        heartbeat_started.set()
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(adapter, "_polling_heartbeat_loop", heartbeat)
+    monkeypatch.setenv("TELEGRAM_WEBHOOK_URL", "https://example.test/telegram")
+    monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", "test-secret")
+
+    assert await adapter.connect() is True
+    assert adapter._webhook_mode is True
+    assert adapter._polling_heartbeat_task is None
+    await adapter.disconnect()
+
+    monkeypatch.delenv("TELEGRAM_WEBHOOK_URL")
+    monkeypatch.delenv("TELEGRAM_WEBHOOK_SECRET")
+    try:
+        assert await adapter.connect(is_reconnect=True) is True
+        assert adapter._webhook_mode is False
+        assert adapter._polling_heartbeat_task is not None
+        assert not adapter._polling_heartbeat_task.done()
+        await asyncio.wait_for(heartbeat_started.wait(), timeout=1)
+        assert heartbeat_modes == [False]
+        assert adapter._send_path_degraded is True
+
+        generation = adapter._polling_generation
+        polling_request = builders[1].polling_request
+        polling_request.result = (200, b'{"ok":true,"result":[]}')
+        await _request_for_generation(generation, polling_request, "getUpdates")
+        await asyncio.wait_for(adapter._polling_progress_verifier_task, timeout=1)
+        assert adapter._send_path_degraded is False
+    finally:
+        await adapter.disconnect()
+
+
[redacted:email]
+async def test_current_polling_generation_success_records_progress():
+    adapter = _make_adapter()
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 3
+    request = _ControlledRequest(result=(200, b'{"ok":true,"result":[]}'))
+
+    instrumented = adapter._instrument_polling_request(request)
+    result = await _request_for_generation(
+        generation, instrumented, "https://api.telegram.org/getUpdates"
+    )
+
+    assert instrumented is request
+    assert result == (200, b'{"ok":true,"result":[]}')
+    assert progress.is_set()
+    assert adapter._polling_network_error_count == 0
+    assert adapter._send_path_degraded is False
+    assert generation > 0
+
+
[redacted:email]
[redacted:email]("error_type", [RuntimeError, asyncio.CancelledError])
+async def test_unsuccessful_polling_request_does_not_record_progress(error_type):
+    adapter = _make_adapter()
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 3
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(error=error_type("request did not complete"))
+    )
+
+    with pytest.raises(error_type):
+        await _request_for_generation(
+            generation, request, "https://api.telegram.org/getUpdates"
+        )
+
+    assert not progress.is_set()
+    assert adapter._polling_network_error_count == 3
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_http_error_response_does_not_record_polling_progress():
+    adapter = _make_adapter()
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 3
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(result=(500, b"bad"))
+    )
+
+    result = await _request_for_generation(
+        generation, request, "https://api.telegram.org/getUpdates"
+    )
+
+    assert result == (500, b"bad")
+    assert not progress.is_set()
+    assert adapter._polling_network_error_count == 3
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_general_request_success_cannot_record_polling_progress(monkeypatch):
+    class _StopConnect(Exception):
+        pass
+
+    class _Builder:
+        def __init__(self):
+            self.general_request = None
+            self.polling_request = None
+
+        def token(self, _token):
+            return self
+
+        def request(self, request):
+            self.general_request = request
+            return self
+
+        def get_updates_request(self, request):
+            self.polling_request = request
+            return self
+
+        def build(self):
+            raise _StopConnect
+
+    builder = _Builder()
+
+    class _Application:
+        @staticmethod
+        def builder():
+            return builder
+
+    _ControlledRequest.instances = []
+
+    async def _no_fallback_ips():
+        return []
+
+    monkeypatch.setattr(tg_adapter, "Application", _Application)
+    monkeypatch.setattr(tg_adapter, "HTTPXRequest", _ControlledRequest)
+    monkeypatch.setattr(tg_adapter, "discover_fallback_ips", _no_fallback_ips)
+    monkeypatch.setattr(tg_adapter, "resolve_proxy_url", lambda *args, **kwargs: None)
+
+    adapter = _make_adapter()
+    monkeypatch.setattr(adapter, "_acquire_platform_lock", lambda *args, **kwargs: True)
+    monkeypatch.setattr(adapter, "_fallback_ips", lambda: [])
+    _, progress = adapter._begin_polling_generation()
+
+    assert await adapter.connect() is False
+    assert builder.general_request is _ControlledRequest.instances[0]
+    assert builder.polling_request is _ControlledRequest.instances[1]
+
+    builder.general_request.result = (200, b'{"ok":true}')
+    result = await builder.general_request.do_request("https://api.telegram.org/sendMessage")
+
+    assert result == (200, b'{"ok":true}')
+    assert not progress.is_set()
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_late_previous_generation_completion_cannot_heal_current_generation():
+    adapter = _make_adapter()
+    generation_1, _ = adapter._begin_polling_generation()
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(
+            result=(200, b'{"ok":true,"result":[]}'),
+            entered=entered,
+            release=release,
+        )
+    )
+
+    completion = asyncio.create_task(
+        _request_for_generation(generation_1, request, "getUpdates")
+    )
+    await entered.wait()
+    generation_2, progress_2 = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 4
+    release.set()
+
+    assert await completion == (200, b'{"ok":true,"result":[]}')
+    assert generation_2 == generation_1 + 1
+    assert not progress_2.is_set()
+    assert adapter._polling_network_error_count == 4
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_old_polling_child_keeps_generation_when_request_entry_is_delayed():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    release_old_request = asyncio.Event()
+    start_count = 0
+    old_child = None
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(result=(200, b'{"ok":true,"result":[]}'))
+    )
+
+    async def start_polling(**_kwargs):
+        nonlocal start_count, old_child
+        start_count += 1
+        if start_count == 1:
+
+            async def delayed_old_request():
+                await release_old_request.wait()
+                return await request.do_request("getUpdates")
+
+            old_child = asyncio.create_task(delayed_old_request())
+
+    adapter._app.updater.start_polling = start_polling
+
+    await adapter._start_polling_once(
+        adapter._app,
+        drop_pending_updates=False,
+        error_callback=MagicMock(),
+    )
+    generation_1 = adapter._polling_generation
+    await adapter._start_polling_once(
+        adapter._app,
+        drop_pending_updates=False,
+        error_callback=MagicMock(),
+    )
+    generation_2 = adapter._polling_generation
+    progress_2 = adapter._polling_progress_event
+    verifier_2 = adapter._polling_progress_verifier_task
+
+    try:
+        release_old_request.set()
+        assert await old_child == (200, b'{"ok":true,"result":[]}')
+
+        assert generation_2 == generation_1 + 1
+        assert not progress_2.is_set()
+        assert adapter._send_path_degraded is True
+    finally:
+        await _cancel_task(verifier_2)
+
+
[redacted:email]
+async def test_error_callback_is_bound_to_its_polling_generation():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    callbacks = []
+    delegated = MagicMock()
+
+    async def capture_start(**kwargs):
+        callbacks.append(kwargs["error_callback"])
+
+    adapter._app.updater.start_polling = capture_start
+    await adapter._start_polling_once(
+        adapter._app,
+        drop_pending_updates=False,
+        error_callback=delegated,
+    )
+    await adapter._start_polling_once(
+        adapter._app,
+        drop_pending_updates=False,
+        error_callback=delegated,
+    )
+    verifier = adapter._polling_progress_verifier_task
+    stale_error = ConnectionError("stale generation")
+    current_error = ConnectionError("current generation")
+
+    try:
+        callbacks[0](stale_error)
+        delegated.assert_not_called()
+
+        callbacks[1](current_error)
+        delegated.assert_called_once_with(current_error)
+    finally:
+        await _cancel_task(verifier)
+
+
[redacted:email]
+async def test_cold_start_waits_for_get_updates_progress_before_healing():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+
+    started = await adapter._start_polling_resilient(
+        drop_pending_updates=True,
+        error_callback=MagicMock(),
+    )
+
+    verifier = adapter._polling_progress_verifier_task
+    assert started is True
+    assert adapter._app.updater.running is True
+    assert adapter._send_path_degraded is True
+    assert verifier is not None and not verifier.done()
+    assert verifier in adapter._background_tasks
+    assert [task for task in adapter._background_tasks if not task.done()] == [verifier]
+    await _cancel_task(verifier)
+
+
[redacted:email]
+async def test_matching_get_updates_progress_heals_and_stops_verifier(monkeypatch):
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    recovery = MagicMock()
+    monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery)
+
+    await adapter._start_polling_resilient(
+        drop_pending_updates=False,
+        error_callback=MagicMock(),
+    )
+    verifier = adapter._polling_progress_verifier_task
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(result=(200, b'{"ok":true,"result":[]}'))
+    )
+    await _request_for_generation(
+        adapter._polling_generation, request, "getUpdates"
+    )
+    await asyncio.wait_for(verifier, timeout=1)
+
+    assert adapter._send_path_degraded is False
+    assert verifier.done()
+    recovery.assert_not_called()
+
+
[redacted:email]
+async def test_general_path_success_without_get_updates_progress_recovers_once(monkeypatch):
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    recovery = MagicMock()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0.01, raising=False)
+    monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery)
+
+    await adapter._start_polling_resilient(
+        drop_pending_updates=False,
+        error_callback=MagicMock(),
+    )
+    verifier = adapter._polling_progress_verifier_task
+    await asyncio.wait_for(verifier, timeout=1)
+
+    assert adapter._app.bot.get_me.await_count == 1
+    recovery.assert_called_once()
+    error = recovery.call_args.args[0]
+    assert isinstance(error, RuntimeError)
+    assert str(error) == "getUpdates made no progress before verifier deadline"
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
[redacted:email](
+    ("probe_error", "should_recover"),
+    [
+        (ConnectionError("pool wedged"), True),
+        (type("InvalidToken", (Exception,), {})("token revoked"), False),
+    ],
+)
+async def test_general_path_error_only_recovers_connectivity_failures(
+    monkeypatch, probe_error, should_recover
+):
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app(get_me=AsyncMock(side_effect=probe_error))
+    recovery = MagicMock()
+    monkeypatch.setattr(tg_adapter, "_POLLING_PROGRESS_TIMEOUT", 0.01, raising=False)
+    monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery)
+
+    await adapter._start_polling_resilient(
+        drop_pending_updates=False,
+        error_callback=MagicMock(),
+    )
+    await asyncio.wait_for(adapter._polling_progress_verifier_task, timeout=1)
+
+    assert recovery.called is should_recover
+    if should_recover:
+        assert recovery.call_args.args[0] is probe_error
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
[redacted:email]("retry_kind", ["network", "conflict"])
+async def test_retry_start_requires_matching_progress_to_heal(monkeypatch, retry_kind):
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    adapter._polling_error_callback_ref = MagicMock()
+    adapter._polling_network_error_count = 3
+    monkeypatch.setattr(tg_adapter.asyncio, "sleep", AsyncMock())
+
+    if retry_kind == "network":
+        await adapter._handle_polling_network_error(ConnectionError("offline"))
+        assert adapter._polling_network_error_count == 4
+    else:
+        await adapter._handle_polling_conflict(
+            RuntimeError("Conflict: terminated by other getUpdates request")
+        )
+        assert adapter._polling_network_error_count == 3
+        assert adapter._polling_conflict_count == 1
+
+    generation = adapter._polling_generation
+    verifier = adapter._polling_progress_verifier_task
+    assert generation > 0
+    assert verifier is not None and not verifier.done()
+    assert adapter._send_path_degraded is True
+
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(result=(200, b'{"ok":true,"result":[]}'))
+    )
+    await _request_for_generation(generation, request, "getUpdates")
+    await asyncio.wait_for(verifier, timeout=1)
+    assert adapter._polling_network_error_count == 0
+    assert adapter._polling_conflict_count == 0
+    assert adapter._send_path_degraded is False
+
+
[redacted:email]
+async def test_repeated_starts_replace_verifier_and_stale_verifier_cannot_heal():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+
+    await adapter._start_polling_resilient(
+        drop_pending_updates=False, error_callback=MagicMock()
+    )
+    generation_1 = adapter._polling_generation
+    progress_1 = adapter._polling_progress_event
+    verifier_1 = adapter._polling_progress_verifier_task
+
+    await adapter._start_polling_resilient(
+        drop_pending_updates=False, error_callback=MagicMock()
+    )
+    verifier_2 = adapter._polling_progress_verifier_task
+    await asyncio.sleep(0)
+
+    assert adapter._polling_generation == generation_1 + 1
+    assert verifier_1.cancelled()
+    assert verifier_2 is not verifier_1 and not verifier_2.done()
+    assert [task for task in adapter._background_tasks if not task.done()] == [verifier_2]
+
+    progress_1.set()
+    await asyncio.sleep(0)
+    assert adapter._send_path_degraded is True
+    await _cancel_task(verifier_2)
+
+
[redacted:email]
+async def test_disconnect_fences_verifier_and_late_progress_completion():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    await adapter._start_polling_resilient(
+        drop_pending_updates=False, error_callback=MagicMock()
+    )
+    generation = adapter._polling_generation
+    verifier = adapter._polling_progress_verifier_task
+
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    request = adapter._instrument_polling_request(
+        _ControlledRequest(
+            result=(200, b'{"ok":true,"result":[]}'),
+            entered=entered,
+            release=release,
+        )
+    )
+    completion = asyncio.create_task(
+        _request_for_generation(generation, request, "getUpdates")
+    )
+    await entered.wait()
+
+    await adapter.disconnect()
+
+    assert verifier.done()
+    assert adapter._polling_progress_verifier_task is None
+    assert adapter._polling_progress_accepting is False
+    assert adapter._polling_generation > generation
+    assert adapter._send_path_degraded is True
+
+    release.set()
+    assert await completion == (200, b'{"ok":true,"result":[]}')
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_disconnect_during_polling_start_returns_false_without_recovery():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    recovery = MagicMock()
+    adapter._schedule_polling_recovery = recovery
+
+    async def blocked_start(**_kwargs):
+        entered.set()
+        await release.wait()
+
+    adapter._app.updater.start_polling = blocked_start
+    start = asyncio.create_task(
+        adapter._start_polling_resilient(
+            drop_pending_updates=False,
+            error_callback=MagicMock(),
+        )
+    )
+    await entered.wait()
+
+    await adapter.disconnect()
+    release.set()
+    result = await start
+
+    assert result is False
+    recovery.assert_not_called()
+    assert adapter._polling_error_task is None
+    assert not adapter.has_fatal_error
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_caller_cancellation_during_polling_start_still_propagates():
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    entered = asyncio.Event()
+    release = asyncio.Event()
+    recovery = MagicMock()
+    adapter._schedule_polling_recovery = recovery
+
+    async def blocked_start(**_kwargs):
+        entered.set()
+        await release.wait()
+
+    adapter._app.updater.start_polling = blocked_start
+    start = asyncio.create_task(
+        adapter._start_polling_resilient(
+            drop_pending_updates=False,
+            error_callback=MagicMock(),
+        )
+    )
+    await entered.wait()
+
+    start.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await start
+
+    assert start.cancelled()
+    recovery.assert_not_called()
+    assert not adapter.has_fatal_error
+    assert adapter._send_path_degraded is True
+    await adapter.disconnect()
+
+
[redacted:email]
+async def test_disconnect_cancels_recovery_before_it_can_rearm_progress(monkeypatch):
+    adapter = _make_adapter()
+    adapter._app = _mock_polling_app()
+    adapter._app.updater.running = False
+    adapter._polling_error_callback_ref = MagicMock()
+
+    drain_entered = asyncio.Event()
+    release_drain = asyncio.Event()
+    start_entered = asyncio.Event()
+    release_start = asyncio.Event()
+    teardown_paused = asyncio.Event()
+    release_teardown = asyncio.Event()
+
+    async def immediate_backoff(_delay):
+        return None
+
+    async def blocked_drain():
+        drain_entered.set()
+        await release_drain.wait()
+
+    async def blocked_start_polling(**_kwargs):
+        start_entered.set()
+        await release_start.wait()
+
+    async def blocked_status_indicator(*, online):
+        assert online is False
+        teardown_paused.set()
+        await release_teardown.wait()
+
+    monkeypatch.setattr(tg_adapter.asyncio, "sleep", immediate_backoff)
+    monkeypatch.setattr(adapter, "_drain_polling_connections", blocked_drain)
+    monkeypatch.setattr(
+        adapter._app.updater, "start_polling", blocked_start_polling
+    )
+    monkeypatch.setattr(adapter, "_set_status_indicator", blocked_status_indicator)
+
+    recovery = asyncio.create_task(
+        adapter._handle_polling_network_error(ConnectionError("offline"))
+    )
+    adapter._polling_error_task = recovery
+    await drain_entered.wait()
+
+    disconnect = asyncio.create_task(adapter.disconnect())
+    await teardown_paused.wait()
+
+    try:
+        # Before the fix, disconnect pauses here before cancelling recovery.
+        # Releasing the recovery lets it begin a fresh generation after the
+        # teardown fence, and matching progress can then heal the adapter.
+        if not recovery.done():
+            release_drain.set()
+            await start_entered.wait()
+
+        rearmed_after_fence = adapter._polling_progress_accepting
+        adapter._record_polling_progress(adapter._polling_generation)
+
+        assert rearmed_after_fence is False
+        assert getattr(adapter, "_polling_teardown_started", False) is True
+        assert adapter._polling_progress_accepting is False
+        assert adapter._send_path_degraded is True
+        assert recovery.done()
+    finally:
+        release_drain.set()
+        release_start.set()
+        release_teardown.set()
+        for task in (recovery, disconnect):
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(recovery, disconnect, return_exceptions=True)

tests/gateway/test_telegram_send_path_health.py

@@ -65,49 +65,32 @@ async def test_send_short_circuits_when_path_degraded():
 
 
 @pytest.mark.asyncio
-async def test_reconnect_storm_sets_and_heartbeat_clears_flag(monkeypatch):
-    """_handle_polling_network_error sets the flag while reconnecting; if the
-    reconnect attempt itself raises (polling not yet healthy), the flag stays
-    True until a later successful heartbeat probe in
-    _verify_polling_after_reconnect clears it."""
+async def test_get_me_success_without_polling_progress_does_not_heal(monkeypatch):
+    """A responsive general Bot API path is not proof that getUpdates works."""
     adapter = _make_adapter()
     adapter._app = MagicMock()
     adapter._app.updater = MagicMock()
     adapter._app.updater.running = True
-    adapter._app.updater.stop = AsyncMock()
-    # First start_polling attempt fails — the reconnect handler must leave the
-    # flag set (path still unhealthy) and not clear it prematurely.
-    adapter._app.updater.start_polling = AsyncMock(side_effect=OSError("still down"))
     adapter._app.bot = MagicMock()
     adapter._app.bot.get_me = AsyncMock(return_value=MagicMock())
-    adapter._polling_error_callback_ref = AsyncMock()
-    monkeypatch.setattr(
-        "plugins.platforms.telegram.adapter.Update", MagicMock(ALL_TYPES=[])
-    )
-    # Suppress the self-rescheduled retry so the test doesn't recurse.
+
+    generation, progress = adapter._begin_polling_generation()
+    recovery = MagicMock()
+    monkeypatch.setattr(adapter, "_schedule_polling_recovery", recovery)
     monkeypatch.setattr(
-        "plugins.platforms.telegram.adapter.asyncio.ensure_future", MagicMock()
+        "plugins.platforms.telegram.adapter._POLLING_PROGRESS_TIMEOUT", 0,
+        raising=False,
     )
+    await adapter._verify_polling_after_reconnect(generation, progress)
 
-    with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock):
-        await adapter._handle_polling_network_error(OSError("Bad Gateway"))
-    # start_polling failed → path still degraded.
+    adapter._app.bot.get_me.assert_awaited_once()
+    recovery.assert_called_once()
     assert adapter._send_path_degraded is True
 
-    # Now the deferred probe runs against a recovered (running) updater and
-    # a responsive bot — it clears the flag.
-    adapter._app.updater.running = True
-    with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock):
-        await adapter._verify_polling_after_reconnect()
-    assert adapter._send_path_degraded is False
-
 
 @pytest.mark.asyncio
-async def test_successful_reconnect_clears_flag_without_probe(monkeypatch):
-    """Regression for #35205: a successful start_polling() clears
-    _send_path_degraded immediately, so outbound sends are not blocked for
-    the full HEARTBEAT_PROBE_DELAY window (and never get stuck True if the
-    deferred probe is never scheduled / never runs)."""
+async def test_successful_reconnect_waits_for_get_updates_progress(monkeypatch):
+    """start_polling() return alone cannot heal; matching progress can."""
     adapter = _make_adapter()
     adapter._app = MagicMock()
     adapter._app.updater = MagicMock()
@@ -120,17 +103,18 @@ async def test_successful_reconnect_clears_flag_without_probe(monkeypatch):
     monkeypatch.setattr(
         "plugins.platforms.telegram.adapter.Update", MagicMock(ALL_TYPES=[])
     )
-    # Don't let the deferred probe run — prove the clear happens in the
-    # reconnect handler itself, not in _verify_polling_after_reconnect.
-    monkeypatch.setattr(
-        adapter, "_verify_polling_after_reconnect", AsyncMock()
-    )
-
     with patch("plugins.platforms.telegram.adapter.asyncio.sleep", new_callable=AsyncMock):
         await adapter._handle_polling_network_error(OSError("Bad Gateway"))
 
+    verifier = adapter._polling_progress_verifier_task
+    assert adapter._send_path_degraded is True
+    assert adapter._polling_network_error_count == 1
+    blocked = await adapter.send("123", "hello")
+    assert blocked.success is False
+
+    adapter._record_polling_progress(adapter._polling_generation)
+    await verifier
     assert adapter._send_path_degraded is False
     assert adapter._polling_network_error_count == 0
-    # And send() works again right away.
     result = await adapter.send("123", "hello")
     assert result.success is True

tests/test_telegram_polling_progress_ptb.py

@@ -0,0 +1,294 @@
+"""Integration coverage for polling progress against the installed PTB runtime."""
+
+import asyncio
+
+import pytest
+
+pytest.importorskip("telegram")  # PTB is an optional dependency; skip if absent.
+from telegram.error import Conflict, TelegramError
+from telegram.request import BaseRequest
+
+from gateway.config import PlatformConfig
+from plugins.platforms.telegram import adapter as tg_adapter
+from plugins.platforms.telegram.adapter import TelegramAdapter
+
+
+class _GeneralRequest(BaseRequest):
+    @property
+    def read_timeout(self):
+        return 10
+
+    async def initialize(self):
+        return None
+
+    async def shutdown(self):
+        return None
+
+    async def do_request(self, url, method, request_data=None, **_kwargs):
+        if url.endswith("/getMe"):
+            return (
+                200,
+                b'{"ok":true,"result":{"id":1,"is_bot":true,'
+                b'"first_name":"Test","username":"test_bot"}}',
+            )
+        return 200, b'{"ok":true,"result":true}'
+
+
+class _GetUpdatesRequest(BaseRequest):
+    def __init__(self):
+        self.initial_conflict_sent = False
+        self.replacement_enabled = False
+        self.replacement_progress_sent = False
+        self.cleanup_calls = 0
+        self.block = asyncio.Event()
+
+    @property
+    def read_timeout(self):
+        return 10
+
+    async def initialize(self):
+        return None
+
+    async def shutdown(self):
+        return None
+
+    async def do_request(self, url, method, request_data=None, **_kwargs):
+        parameters = request_data.parameters if request_data is not None else {}
+        timeout = parameters.get("timeout")
+        timeout_seconds = (
+            timeout.total_seconds() if hasattr(timeout, "total_seconds") else timeout
+        )
+        if timeout_seconds == 0:
+            self.cleanup_calls += 1
+            return 200, b'{"ok":true,"result":[]}'
+        if not self.initial_conflict_sent:
+            self.initial_conflict_sent = True
+            return (
+                409,
+                b'{"ok":false,"error_code":409,'
+                b'"description":"Conflict: another getUpdates request"}',
+            )
+        if self.replacement_enabled and not self.replacement_progress_sent:
+            self.replacement_progress_sent = True
+            return 200, b'{"ok":true,"result":[]}'
+        await self.block.wait()
+        return 200, b'{"ok":true,"result":[]}'
+
+
+class _EnvelopeRequest(BaseRequest):
+    def __init__(self, payload):
+        self.payload = payload
+
+    @property
+    def read_timeout(self):
+        return 10
+
+    async def initialize(self):
+        return None
+
+    async def shutdown(self):
+        return None
+
+    async def do_request(self, url, method, request_data=None, **_kwargs):
+        return 200, self.payload
+
+
+async def _cancel_task(task):
+    if task is None or task.done():
+        return
+    task.cancel()
+    await asyncio.gather(task, return_exceptions=True)
+
+
[redacted:email]
+async def test_real_base_request_invalid_200_body_cannot_record_progress():
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 4
+    adapter._polling_conflict_count = 3
+    request = adapter._instrument_polling_request(_EnvelopeRequest(b"not-json"))
+    context_token = [redacted:secret-assignment]
+
+    try:
+        with pytest.raises(TelegramError, match="Invalid server response"):
+            await request.post("https://api.telegram.org/bot-token/getUpdates")
+    finally:
+        tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token)
+
+    assert not progress.is_set()
+    assert adapter._polling_network_error_count == 4
+    assert adapter._polling_conflict_count == 3
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_real_base_request_bom_rejected_by_ptb_cannot_record_progress():
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 4
+    adapter._polling_conflict_count = 3
+    request = adapter._instrument_polling_request(
+        _EnvelopeRequest(b'\xef\xbb\xbf{"ok":true,"result":[]}')
+    )
+    context_token = [redacted:secret-assignment]
+
+    try:
+        with pytest.raises(TelegramError, match="Invalid server response"):
+            await request.post("https://api.telegram.org/bot-token/getUpdates")
+    finally:
+        tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token)
+
+    assert not progress.is_set()
+    assert adapter._polling_network_error_count == 4
+    assert adapter._polling_conflict_count == 3
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_real_base_request_ptb_replacement_decode_records_progress():
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 4
+    adapter._polling_conflict_count = 3
+    request = adapter._instrument_polling_request(
+        _EnvelopeRequest(b'{"ok":true,"result":[],"note":"\xff"}')
+    )
+    context_token = [redacted:secret-assignment]
+
+    try:
+        result = await request.post("https://api.telegram.org/bot-token/getUpdates")
+    finally:
+        tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token)
+
+    assert result == []
+    assert progress.is_set()
+    assert adapter._polling_network_error_count == 0
+    assert adapter._polling_conflict_count == 0
+    assert adapter._send_path_degraded is False
+
+
[redacted:email]
[redacted:email](
+    ("payload", "missing_result"),
+    [
+        (b'{"ok":false,"result":[]}', False),
+        (b'{"ok":true}', True),
+    ],
+)
+async def test_real_base_request_unsuccessful_200_envelope_cannot_record_progress(
+    payload, missing_result
+):
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 4
+    adapter._polling_conflict_count = 3
+    request = adapter._instrument_polling_request(_EnvelopeRequest(payload))
+    context_token = [redacted:secret-assignment]
+
+    try:
+        if missing_result:
+            with pytest.raises(KeyError, match="result"):
+                await request.post("https://api.telegram.org/bot-token/getUpdates")
+        else:
+            assert await request.post(
+                "https://api.telegram.org/bot-token/getUpdates"
+            ) == []
+    finally:
+        tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token)
+
+    assert not progress.is_set()
+    assert adapter._polling_network_error_count == 4
+    assert adapter._polling_conflict_count == 3
+    assert adapter._send_path_degraded is True
+
+
[redacted:email]
+async def test_real_base_request_valid_success_envelope_records_progress():
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+    generation, progress = adapter._begin_polling_generation()
+    adapter._polling_network_error_count = 4
+    adapter._polling_conflict_count = 3
+    request = adapter._instrument_polling_request(
+        _EnvelopeRequest(b'{"ok":true,"result":[]}')
+    )
+    context_token = [redacted:secret-assignment]
+
+    try:
+        result = await request.post(
+            "https://api.telegram.org/bot-token/getUpdates"
+        )
+    finally:
+        tg_adapter._POLLING_GENERATION_CONTEXT.reset(context_token)
+
+    assert result == []
+    assert progress.is_set()
+    assert adapter._polling_network_error_count == 0
+    assert adapter._polling_conflict_count == 0
+    assert adapter._send_path_degraded is False
+
+
[redacted:email]
+async def test_real_ptb_stop_cleanup_cannot_heal_recovery_generation():
+    assert tg_adapter.TELEGRAM_AVAILABLE is True
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
+    polling_request = _GetUpdatesRequest()
+    app = (
+        tg_adapter.Application.builder()
+        .token("123456:test-token")
+        .request(_GeneralRequest())
+        .get_updates_request(adapter._instrument_polling_request(polling_request))
+        .build()
+    )
+    adapter._app = app
+    adapter._polling_network_error_count = 4
+    adapter._polling_conflict_count = 3
+    callback_called = asyncio.Event()
+    recovery_task = None
+
+    async def stop_for_recovery():
+        await app.updater.stop()
+
+    def schedule_recovery(error):
+        nonlocal recovery_task
+        assert isinstance(error, Conflict)
+        recovery_task = asyncio.create_task(stop_for_recovery())
+        callback_called.set()
+
+    await app.initialize()
+    try:
+        await adapter._start_polling_once(
+            app,
+            drop_pending_updates=False,
+            error_callback=schedule_recovery,
+        )
+        generation = adapter._polling_generation
+        progress = adapter._polling_progress_event
+        await asyncio.wait_for(callback_called.wait(), timeout=2)
+        await asyncio.wait_for(recovery_task, timeout=3)
+
+        assert polling_request.cleanup_calls == 1
+        assert not progress.is_set()
+        assert adapter._polling_network_error_count == 4
+        assert adapter._polling_conflict_count == 3
+        assert adapter._send_path_degraded is True
+
+        polling_request.replacement_enabled = True
+        await adapter._start_polling_once(
+            app,
+            drop_pending_updates=False,
+            error_callback=schedule_recovery,
+        )
+        replacement_generation = adapter._polling_generation
+        replacement_progress = adapter._polling_progress_event
+        await asyncio.wait_for(replacement_progress.wait(), timeout=2)
+
+        assert replacement_generation == generation + 1
+        assert adapter._polling_network_error_count == 0
+        assert adapter._polling_conflict_count == 0
+        assert adapter._send_path_degraded is False
+    finally:
+        polling_request.block.set()
+        if app.updater.running:
+            await app.updater.stop()
+        await _cancel_task(adapter._polling_progress_verifier_task)
+        await app.shutdown()

#64368 fix(telegram): classify and dedup post-reconnect probe failures (#63243)

merged to main

files: plugins/platforms/telegram/adapter.py, tests/gateway/test_telegram_network_reconnect.py

2 PRs closed without merging

From the close comments: merged via another change 2

#63247 fix(telegram): classify and dedup post-reconnect probe failures (#63243) · merged via another change → #64368, #62098
#67913 fix(telegram): serialize post-reconnect recovery · merged via another change → #64368, #67913

Closed issues · 2

Closed#63243bug(telegram): post-reconnect probe routes auth errors into reconnect ladder and bypasses the recovery dedup guard (gap left by #62098)
Closed#67912[Bug]: Telegram post-reconnect probe bypasses recovery single-flight guard

How this cluster was grouped

Link evidence (structural links throughout): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
Pairwise link density 0.36: the share of item pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. 1.00 means every pair is linked.

Pick up a Lab task

Check whether a fix covers the open issues here, with your own agent. Bring back evidence for maintainers.

Open the Lab

At a glance

Merged PRs: #64361, #64368
Closed issues: 2
Attempts: 2 PRs closed without merging
Fix link checked: 14 Jul 2026

For your agent

The same facts, typed.

cluster/63243.json
cluster/63243.md