# NousResearch/hermes-agent · Telegram post-reconnect probe recovery bug fixes

## Open now

- Nothing open on GitHub.

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.

## TL;DR

- 6 member(s): 0 open issue(s), 2 closed issue(s), 0 open PR(s), 2 merged PR(s), 2 PR(s) closed without merging.
- A fix was merged to main in PR #64361. Verify it addresses your case.
- **Recommended:** read PR #64361 (and any competing PRs below) before filing. Confirm the merged change covers your symptom.

## Maintainer resolution verdict

- Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-64361`. This is distinct from the fix status above. Weigh it yourself.

## How this cluster was grouped

- Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference.
- Pairwise link density `0.36`: the share of member 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.

## Closed / duplicate issues

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

## Prior & competing PRs

- **#64361: fix(telegram): require getUpdates progress before polling is healthy**. merged to main · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - 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`
  - fix hunks:
  - `plugins/platforms/telegram/adapter.py`

    ```diff
    @@ -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
     
    ```

    ```diff
    @@ -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):
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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:
    ```

    ```diff
    @@ -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():
    ```

    ```diff
    @@ -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:
    ```

    ```diff
    @@ -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 "
    ```

    ```diff
    @@ -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
     
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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)
                     )
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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:
    ```

    ```diff
    @@ -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",
    ```

    ```diff
    @@ -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",
    ```

    ```diff
    @@ -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.
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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.",
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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 = (
    ```

    ```diff
    @@ -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",
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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,
    ```

    ```diff
    @@ -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):
    ```

    ```diff
    @@ -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()
    ```

    ```diff
    @@ -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."""
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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`

    ```diff
    @@ -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`

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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`

    ```diff
    @@ -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
     
     
    ```

    ```diff
    @@ -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():
         """
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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()]
    ```

    ```diff
    @@ -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
     
     
    ```

    ```diff
    @@ -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
     
     
    ```

    ```diff
    @@ -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()
     
    ```

    ```diff
    @@ -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()
     
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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.
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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.
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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
    ```

    ```diff
    @@ -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.
    ```

    ```diff
    @@ -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()
    ```

    ```diff
    @@ -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`

    ```diff
    @@ -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`

    ```diff
    @@ -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()
    ```

    ```diff
    @@ -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`

    ```diff
    @@ -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 · fix tier `verified`
  - maintainer direction: category `merged`; implication `superseded`
  - files: `plugins/platforms/telegram/adapter.py`, `tests/gateway/test_telegram_network_reconnect.py`
- **#63247: fix(telegram): classify and dedup post-reconnect probe failures (#63243)**. closed without merge
  - maintainer direction: category `merged_via`; implication `superseded`; refs `#64368`, `#62098`
- **#67913: fix(telegram): serialize post-reconnect recovery**. closed without merge
  - maintainer direction: category `merged_via`; implication `superseded`; refs `#64368`, `#64368`, `#67913`, `#64368`

## Fix status

- Merged to main in PR #64361 (cited above). No release data is tracked. Verify downstream.
- Fix last verified (merged) `2026-07-14`.