Telegram 409 polling conflict loop fixes and gateway recovery

Fixes on record

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

• #2312 fix(gateway): retry Telegram 409 polling conflicts before giving up

  merged to main

  files: gateway/platforms/telegram.py, tests/gateway/test_telegram_conflict.py

  Show the diff

  • gateway/platforms/telegram.py

    @@ -129,6 +129,8 @@ def __init__(self, config: PlatformConfig):
             self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
             self._token_lock_identity: [redacted:secret-assignment] = None
             self._polling_error_task: Optional[asyncio.Task] = None
    +        self._polling_conflict_count: int = 0
    +        self._polling_error_callback_ref = None
     
         @staticmethod
         def _looks_like_polling_conflict(error: Exception) -> bool:
    @@ -142,10 +144,49 @@ def _looks_like_polling_conflict(error: Exception) -> bool:
         async def _handle_polling_conflict(self, error: Exception) -> None:
             if self.has_fatal_error and self.fatal_error_code == "telegram_polling_conflict":
                 return
    +        # Track consecutive conflicts — transient 409s can occur when a
    +        # previous gateway instance hasn't fully released its long-poll
    +        # session on Telegram's server (e.g. during --replace handoffs or
    +        # systemd Restart=on-failure respawns).  Retry a few times before
    +        # giving up, so the old session has time to expire.
    +        self._polling_conflict_count += 1
    +
    +        MAX_CONFLICT_RETRIES = 3
    +        RETRY_DELAY = 10  # seconds
    +
    +        if self._polling_conflict_count <= MAX_CONFLICT_RETRIES:
    +            logger.warning(
    +                "[%s] Telegram polling conflict (%d/%d), will retry in %ds. Error: %s",
    +                self.name, self._polling_conflict_count, MAX_CONFLICT_RETRIES,
    +                RETRY_DELAY, error,
    +            )
    +            try:
    +                if self._app and self._app.updater and self._app.updater.running:
    +                    await self._app.updater.stop()
    +            except Exception:
    +                pass
    +            await asyncio.sleep(RETRY_DELAY)
    +            try:
    +                await self._app.updater.start_polling(
    +                    allowed_updates=Update.ALL_TYPES,
    +                    drop_pending_updates=False,
    +                    error_callback=self._polling_error_callback_ref,
    +                )
    +                logger.info("[%s] Telegram polling resumed after conflict retry %d", self.name, self._polling_conflict_count)
    +                self._polling_conflict_count = 0  # reset on success
    +                return
    +            except Exception as retry_err:
    +                logger.warning("[%s] Telegram polling retry failed: %s", self.name, retry_err)
    +                # Don't fall through to fatal yet — wait for the next conflict
    +                # to trigger another retry attempt (up to MAX_CONFLICT_RETRIES).
    +                return
    +
    +        # Exhausted retries — fatal
             message = (
                 "Another Telegram bot poller is already using this token. "
    -            "Hermes stopped Telegram polling to avoid endless retry spam. "
    +            "Hermes stopped Telegram polling after %d retries. "
                 "Make sure only one gateway instance is running for this bot token."
    +            % MAX_CONFLICT_RETRIES
             )
             logger.error("[%s] %s Original error: %s", self.name, message, error)
             self._set_fatal_error("telegram_polling_conflict", message, retryable=False)
    @@ -242,6 +283,9 @@ def _polling_error_callback(error: Exception) -> None:
                         return
                     self._polling_error_task = loop.create_task(self._handle_polling_conflict(error))
     
    +            # Store reference for retry use in _handle_polling_conflict
    +            self._polling_error_callback_ref = _polling_error_callback
    +
                 await self._app.updater.start_polling(
                     allowed_updates=Update.ALL_TYPES,
                     drop_pending_updates=True,

  • tests/gateway/test_telegram_conflict.py

    @@ -47,8 +47,9 @@ async def test_connect_rejects_same_host_token_lock(monkeypatch):
     
     
     @pytest.mark.asyncio
    -async def test_polling_conflict_stops_polling_and_notifies_handler(monkeypatch):
    -    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="[redacted:secret-assignment]"))
    +async def test_polling_conflict_retries_before_fatal(monkeypatch):
    +    """A single 409 should trigger a retry, not an immediate fatal error."""
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
         fatal_handler = AsyncMock()
         adapter.set_fatal_error_handler(fatal_handler)
     
    @@ -69,6 +70,7 @@ async def fake_start_polling(**kwargs):
         updater = SimpleNamespace(
             start_polling=AsyncMock(side_effect=fake_start_polling),
             stop=AsyncMock(),
    +        running=True,
         )
         bot = SimpleNamespace(set_my_commands=AsyncMock())
         app = SimpleNamespace(
    @@ -83,20 +85,102 @@ async def fake_start_polling(**kwargs):
         builder.build.return_value = app
         monkeypatch.setattr("gateway.platforms.telegram.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
     
    +    # Speed up retries for testing
    +    monkeypatch.setattr("asyncio.sleep", AsyncMock())
    +
         ok = await adapter.connect()
     
         assert ok is True
         assert callable(captured["error_callback"])
     
         conflict = type("Conflict", (Exception,), {})
    -    captured["error_callback"](conflict("Conflict: terminated by other getUpdates request; make sure that only one bot instance is running"))
     
    +    # 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)
    +
    +    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"
    +
    +
    [redacted:email]
    +async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
    +    """After exhausting retries, the conflict should become fatal."""
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    +    fatal_handler = AsyncMock()
    +    adapter.set_fatal_error_handler(fatal_handler)
    +
    +    monkeypatch.setattr(
    +        "gateway.status.acquire_scoped_lock",
    +        lambda scope, identity, metadata=None: (True, None),
    +    )
    +    monkeypatch.setattr(
    +        "gateway.status.release_scoped_lock",
    +        lambda scope, identity: None,
    +    )
    +
    +    captured = {}
    +
    +    async def fake_start_polling(**kwargs):
    +        captured["error_callback"] = kwargs["error_callback"]
    +
    +    # Make start_polling fail on retries to exhaust retries
    +    call_count = {"n": 0}
     
    -    assert adapter.fatal_error_code == "telegram_polling_conflict"
    +    async def failing_start_polling(**kwargs):
    +        call_count["n"] += 1
    +        if call_count["n"] == 1:
    +            # First call (initial connect) succeeds
    +            captured["error_callback"] = kwargs["error_callback"]
    +        else:
    +            # Retry calls fail
    +            raise Exception("Connection refused")
    +
    +    updater = SimpleNamespace(
    +        start_polling=AsyncMock(side_effect=failing_start_polling),
    +        stop=AsyncMock(),
    +        running=True,
    +    )
    +    bot = SimpleNamespace(set_my_commands=AsyncMock())
    +    app = SimpleNamespace(
    +        bot=bot,
    +        updater=updater,
    +        add_handler=MagicMock(),
    +        initialize=AsyncMock(),
    +        start=AsyncMock(),
    +    )
    +    builder = MagicMock()
    +    builder.token.return_value = builder
    +    builder.build.return_value = app
    +    monkeypatch.setattr("gateway.platforms.telegram.Application", SimpleNamespace(builder=MagicMock(return_value=builder)))
    +
    +    # Speed up retries for testing
    +    monkeypatch.setattr("asyncio.sleep", AsyncMock())
    +
    +    ok = await adapter.connect()
    +    assert ok is True
    +
    +    conflict = type("Conflict", (Exception,), {})
    +
    +    # Directly call _handle_polling_conflict to avoid event-loop scheduling
    +    # complexity.  Each call simulates one 409 from Telegram.
    +    for i in range(4):
    +        await adapter._handle_polling_conflict(
    +            conflict("Conflict: terminated by other getUpdates request")
    +        )
    +
    +    # After 3 failed retries (count 1-3 each enter the retry branch but
    +    # start_polling raises), the 4th conflict pushes count to 4 which
    +    # exceeds MAX_CONFLICT_RETRIES (3), entering the fatal branch.
    +    assert adapter.fatal_error_code == "telegram_polling_conflict", (
    +        f"Expected fatal after 4 conflicts, got code={adapter.fatal_error_code}, "
    +        f"count={adapter._polling_conflict_count}"
    +    )
         assert adapter.has_fatal_error is True
    -    updater.stop.assert_awaited()
         fatal_handler.assert_awaited_once()
     
     

• #53941 fix(telegram): kill 409 polling conflict loop by disarming PTB retry synchronously

  merged to main

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

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -1696,6 +1696,69 @@ async def _verify_polling_after_reconnect(self) -> None:
                 )
                 await self._handle_polling_network_error(probe_err)
     
    +    def _disarm_ptb_retry_loop(self) -> None:
    +        """Synchronously stop PTB's internal polling retry loop.
    +
    +        PTB wraps ``getUpdates`` in ``network_retry_loop`` with
    +        ``max_retries=-1`` (retry forever).  When a ``TelegramError`` (including
    +        a 409 ``Conflict``) fires, that loop calls our ``error_callback``
    +        *synchronously*, then sleeps and re-checks ``while is_running()`` before
    +        polling again.  Our ``error_callback`` only schedules an async recovery
    +        task (``loop.create_task(...)``) and returns immediately, so PTB's loop
    +        keeps polling while our handler concurrently runs
    +        ``stop -> sleep -> start_polling``.  The two polling sessions overlap and
    +        Telegram returns a fresh 409 — a self-inflicted conflict loop on a
    +        ~31s cadence.
    +
    +        The loop is wired with ``is_running=lambda: updater.running`` and a
    +        private ``stop_event`` (``do_action`` races that event and returns the
    +        moment it is set).  Setting that event *synchronously inside the
    +        callback* — before it returns — makes PTB's loop exit on its own next
    +        tick instead of racing our recovery.  Our async handler then performs
    +        the real ``await updater.stop()`` (idempotent) followed by
    +        drain + ``start_polling()``, which builds a fresh ``stop_event`` so the
    +        restart is not poisoned.
    +
    +        Best-effort and defensive: PTB names the attribute differently across
    +        versions (``_Updater__polling_task_stop_event`` via name-mangling), so
    +        we probe for both spellings.  If neither is found we do nothing and
    +        fall back to the prior behaviour (async ``updater.stop()`` racing PTB) —
    +        i.e. we never make things worse than before.
    +
    +        We deliberately do NOT fall back to flipping ``updater._running``:
    +        ``stop()`` raises ``RuntimeError`` when ``running`` is already False and
    +        our recovery handler guards its ``stop()`` call on ``running``, so
    +        clearing the flag here would skip the real teardown and leave PTB's
    +        stop_event uncleared — poisoning the subsequent ``start_polling()``.
    +        The stop_event lever leaves ``_running`` True, so the handler's
    +        ``await updater.stop()`` still runs, drains the polling task, and clears
    +        the event for a clean restart.
    +        """
    +        updater = getattr(self._app, "updater", None) if self._app else None
    +        if updater is None:
    +            return
    +        # Preferred (and only) lever: PTB's polling stop_event. Name-mangled on
    +        # Updater, so probe both the mangled and unmangled spellings.
    +        for attr in (
    +            "_Updater__polling_task_stop_event",
    +            "_polling_task_stop_event",
    +        ):
    +            stop_event = getattr(updater, attr, None)
    +            if isinstance(stop_event, asyncio.Event):
    +                if not stop_event.is_set():
    +                    stop_event.set()
    +                    logger.debug(
    +                        "[%s] Disarmed PTB polling retry loop via %s",
    +                        self.name, attr,
    +                    )
    +                return
    +        logger.debug(
    +            "[%s] Could not disarm PTB polling retry loop "
    +            "(stop_event not found on this PTB version); "
    +            "falling back to async stop()",
    +            self.name,
    +        )
    +
         async def _handle_polling_conflict(self, error: Exception) -> None:
             if self.has_fatal_error and self.fatal_error_code == "telegram_polling_conflict":
                 return
    @@ -2388,6 +2451,14 @@ def _polling_error_callback(error: Exception) -> None:
                         if self._polling_error_task and not self._polling_error_task.done():
                             return
                         if self._looks_like_polling_conflict(error):
    +                        # Synchronously stop PTB's internal network_retry_loop
    +                        # BEFORE scheduling our async recovery task.  PTB calls
    +                        # this callback synchronously inside its loop and then
    +                        # keeps polling on its own; if we only schedule a task
    +                        # here, PTB's retry and our stop->restart overlap and
    +                        # produce a fresh 409.  Disarming the loop now makes it
    +                        # exit on its next tick so recovery owns polling alone.
    +                        self._disarm_ptb_retry_loop()
                             self._polling_error_task = loop.create_task(self._handle_polling_conflict(error))
                         elif self._looks_like_network_error(error):
                             logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error)

  • tests/gateway/test_telegram_conflict.py

    @@ -489,3 +489,113 @@ async def test_reconnect_preserves_pending_updates(monkeypatch):
         assert ok is True
         assert captured["drop_pending_updates"] is False
         await _cancel_heartbeat(adapter)
    +
    +
    [redacted:email]
    +async def test_disarm_sets_ptb_stop_event():
    +    """_disarm_ptb_retry_loop sets PTB's name-mangled polling stop_event.
    +
    +    This is the root-cause fix for the 409 conflict loop (#30122): the
    +    error_callback must synchronously signal PTB's internal network_retry_loop
    +    to stop BEFORE our async recovery task restarts polling, otherwise the two
    +    polling sessions overlap and produce a fresh 409.
    +    """
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    +
    +    stop_event = asyncio.Event()
    +    # PTB stores it name-mangled as _Updater__polling_task_stop_event.
    +    updater = SimpleNamespace(running=True)
    +    setattr(updater, "_Updater__polling_task_stop_event", stop_event)
    +    adapter._app = SimpleNamespace(updater=updater)
    +
    +    assert not stop_event.is_set()
    +    adapter._disarm_ptb_retry_loop()
    +    assert stop_event.is_set(), "disarm must set PTB's polling stop_event"
    +    # Must not flip _running — the recovery handler's stop() guards on running
    +    # and stop() raises if running is already False.
    +    assert updater.running is True
    +
    +
    [redacted:email]
    +async def test_disarm_noop_when_stop_event_absent():
    +    """When PTB exposes no stop_event, disarm is a safe no-op (no regression).
    +
    +    It must NOT flip _running (which would make the handler skip stop() and
    +    leave the loop wedged) — it just falls back to the prior async stop() race.
    +    """
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    +    updater = SimpleNamespace(running=True, _running=True)
    +    adapter._app = SimpleNamespace(updater=updater)
    +
    +    adapter._disarm_ptb_retry_loop()  # no stop_event attribute present
    +
    +    assert updater.running is True
    +    assert updater._running is True, "disarm must not flip _running as a fallback"
    +
    +
    [redacted:email]
    +async def test_conflict_callback_disarms_before_scheduling(monkeypatch):
    +    """The polling error_callback disarms PTB synchronously, then schedules
    +    recovery — proving the fix is wired into the live callback, not just the
    +    helper (#30122)."""
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    +    fatal_handler = AsyncMock()
    +    adapter.set_fatal_error_handler(fatal_handler)
    +
    +    monkeypatch.setattr(
    +        "gateway.status.acquire_scoped_lock",
    +        lambda scope, identity, metadata=None: (True, None),
    +    )
    +    monkeypatch.setattr(
    +        "gateway.status.release_scoped_lock",
    +        lambda scope, identity: None,
    +    )
    +    monkeypatch.setattr("asyncio.sleep", AsyncMock())
    +
    +    captured = {}
    +
    +    async def fake_start_polling(**kwargs):
    +        captured["error_callback"] = kwargs["error_callback"]
    +
    +    stop_event = asyncio.Event()
    +    updater = SimpleNamespace(
    +        start_polling=AsyncMock(side_effect=fake_start_polling),
    +        stop=AsyncMock(),
    +        running=True,
    +    )
    +    setattr(updater, "_Updater__polling_task_stop_event", stop_event)
    +    bot = SimpleNamespace(set_my_commands=AsyncMock(), delete_webhook=AsyncMock())
    +    app = SimpleNamespace(
    +        bot=bot,
    +        updater=updater,
    +        add_handler=MagicMock(),
    +        initialize=AsyncMock(),
    +        start=AsyncMock(),
    +    )
    +    builder = MagicMock()
    +    builder.token.return_value = builder
    +    builder.request.return_value = builder
    +    builder.get_updates_request.return_value = builder
    +    builder.build.return_value = app
    +    monkeypatch.setattr(
    +        "plugins.platforms.telegram.adapter.Application",
    +        SimpleNamespace(builder=MagicMock(return_value=builder)),
    +    )
    +
    +    ok = await adapter.connect()
    +    assert ok is True
    +
    +    conflict = type("Conflict", (Exception,), {})
    +    # Fire a 409 through the live callback. The disarm must happen
    +    # synchronously (before any await), so the stop_event is set immediately
    +    # on return — before the scheduled recovery task gets a chance to run.
    +    assert not stop_event.is_set()
    +    captured["error_callback"](conflict("Conflict: terminated by other getUpdates"))
    +    assert stop_event.is_set(), "callback must disarm PTB synchronously"
    +    assert adapter._polling_error_task is not None, "recovery task must be scheduled"
    +
    +    # Drain the scheduled recovery task so it doesn't outlive the test.
    +    for _ in range(10):
    +        await asyncio.sleep(0)
    +    await _cancel_heartbeat(adapter)
    +

• #68447 fix(telegram): preserve fatal recovery handoff from self-cancelling teardown

  merged to main

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

  Show the diff

  • plugins/platforms/telegram/adapter.py

    @@ -2357,11 +2357,11 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
             if attempt > MAX_NETWORK_RETRIES:
                 message = (
                     "Telegram polling could not reconnect after %d network error retries. "
    -                "Restarting gateway." % MAX_NETWORK_RETRIES
    +                "Escalating to gateway recovery." % MAX_NETWORK_RETRIES
                 )
                 logger.error("[%s] %s Last error: %s", self.name, message, _redact_telegram_error_text(error))
                 self._set_fatal_error("telegram_network_error", message, retryable=True)
    -            await self._notify_fatal_error()
    +            await self._handoff_polling_fatal_error()
                 return
     
             delay = min(BASE_DELAY * (2 ** (attempt - 1)), MAX_DELAY)
    @@ -2523,7 +2523,7 @@ async def _polling_heartbeat_loop(self) -> None:
                                 "gateway reconnect." % stuck_for,
                                 retryable=True,
                             )
    -                        await self._notify_fatal_error()
    +                        await self._handoff_polling_fatal_error()
                             return
                     else:
                         stuck_task_ref = None
    @@ -2982,7 +2982,24 @@ async def _handle_polling_conflict(self, error: Exception) -> None:
                     self.name, stop_error, exc_info=True,
                 )
             if not _already_fatal:
    -            await self._notify_fatal_error()
    +            await self._handoff_polling_fatal_error()
    +
    +    async def _handoff_polling_fatal_error(self) -> None:
    +        """Notify the runner without letting child teardown cancel this owner.
    +
    +        The runner bounds adapter cleanup in a child task.  ``disconnect()``
    +        cancels the tracked polling-recovery task and the heartbeat task, so
    +        retaining the current notifier in either field would cancel the fatal
    +        callback before the runner can finish its reconnect or shutdown
    +        decision.  Release only the current owner from whichever field tracks
    +        it; unrelated tasks remain under teardown control.
    +        """
    +        current_task = asyncio.current_task()
    +        if self._polling_error_task is current_task:
    +            self._polling_error_task = None
    +        if getattr(self, "_polling_heartbeat_task", None) is current_task:
    +            self._polling_heartbeat_task = None
    +        await self._notify_fatal_error()
     
         async def _create_dm_topic(
             self,

  • tests/gateway/test_telegram_conflict.py

    @@ -293,6 +293,32 @@ async def failing_start_polling(**kwargs):
         await _cancel_heartbeat(adapter)
     
     
    [redacted:email]
    +async def test_conflict_exhaustion_hands_off_before_child_disconnect():
    +    """The conflict recovery owner must survive its fatal callback handoff."""
    +    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    +    adapter._polling_conflict_count = 5  # MAX_CONFLICT_RETRIES
    +    disconnect_tasks = []
    +
    +    async def fatal_handler(failed_adapter):
    +        disconnect_task = asyncio.create_task(failed_adapter.disconnect())
    +        disconnect_tasks.append(disconnect_task)
    +        await asyncio.wait({disconnect_task})
    +
    +    adapter.set_fatal_error_handler(fatal_handler)
    +
    +    conflict = type("Conflict", (Exception,), {})
    +    recovery_task = asyncio.create_task(
    +        adapter._handle_polling_conflict(conflict("getUpdates conflict"))
    +    )
    +    adapter._polling_error_task = recovery_task
    +    result = await asyncio.gather(recovery_task, return_exceptions=True)
    +    await asyncio.gather(*disconnect_tasks, return_exceptions=True)
    +
    +    assert result == [None]
    +    assert adapter._polling_error_task is None
    +
    +
     @pytest.mark.asyncio
     async def test_connect_marks_retryable_fatal_error_for_startup_network_failure(monkeypatch):
         adapter = TelegramAdapter(PlatformConfig(enabled=True, token="***"))
    @@ -730,4 +756,3 @@ async def fake_start_polling(**kwargs):
         for _ in range(10):
             await asyncio.sleep(0)
         await _cancel_heartbeat(adapter)
    -

  • tests/gateway/test_telegram_network_reconnect.py

    @@ -14,7 +14,7 @@
     
     import pytest
     
    -from gateway.config import PlatformConfig
    +from gateway.config import GatewayConfig, Platform, PlatformConfig
     
     
     def _ensure_telegram_mock():
    @@ -37,6 +37,7 @@ def _ensure_telegram_mock():
     
     from plugins.platforms.telegram import adapter as tg_adapter  # noqa: E402
     from plugins.platforms.telegram.adapter import TelegramAdapter  # noqa: E402
    +from gateway.run import GatewayRunner  # noqa: E402
     
     
     @pytest.fixture(autouse=True)
    @@ -224,6 +225,76 @@ async def test_reconnect_triggers_fatal_after_max_retries():
         fatal_handler.assert_called_once()
     
     
    [redacted:email]
    +async def test_retry_exhaustion_queues_reconnect_before_child_disconnect(tmp_path):
    +    """Fatal teardown must not cancel the gateway's reconnect handoff.
    +
    +    The gateway runs ``disconnect()`` in a bounded child task.  If the current
    +    polling-recovery owner remains in ``_polling_error_task``, Telegram teardown
    +    cancels that parent while it is still awaiting the fatal handler, so the
    +    handler never gets to queue background reconnection.
    +    """
    +    config = GatewayConfig(
    +        platforms={
    +            Platform.TELEGRAM: PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +        },
    +        sessions_dir=tmp_path / "sessions",
    +    )
    +    runner = GatewayRunner(config)
    +    adapter = _make_adapter()
    +    adapter._polling_network_error_count = 10  # MAX_NETWORK_RETRIES
    +    adapter.set_fatal_error_handler(runner._handle_adapter_fatal_error)
    +    runner.adapters = {Platform.TELEGRAM: adapter}
    +    runner.delivery_router.adapters = runner.adapters
    +
    +    recovery_task = asyncio.create_task(
    +        adapter._handle_polling_network_error(Exception("still failing"))
    +    )
    +    adapter._polling_error_task = recovery_task
    +    result = await asyncio.gather(recovery_task, return_exceptions=True)
    +
    +    assert result == [None]
    +    assert runner.adapters == {}
    +    assert Platform.TELEGRAM in runner._failed_platforms
    +    assert runner._failed_platforms[Platform.TELEGRAM]["attempts"] == 0
    +
    +
    [redacted:email]
    +async def test_heartbeat_watchdog_handoff_survives_child_disconnect(tmp_path):
    +    """The wedged-recovery heartbeat watchdog must survive its fatal callback.
    +
    +    The heartbeat loop force-escalates a stuck polling-recovery task.  Like
    +    the network/conflict terminal paths, the heartbeat task itself is the
    +    owner that ``disconnect()`` cancels, so the fatal callback must release
    +    ``_polling_heartbeat_task`` before notifying the runner.
    +    """
    +    config = GatewayConfig(
    +        platforms={
    +            Platform.TELEGRAM: PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +        },
    +        sessions_dir=tmp_path / "sessions",
    +    )
    +    runner = GatewayRunner(config)
    +    adapter = _make_adapter()
    +    adapter.set_fatal_error_handler(runner._handle_adapter_fatal_error)
    +    runner.adapters = {Platform.TELEGRAM: adapter}
    +    runner.delivery_router.adapters = runner.adapters
    +
    +    # Simulate the heartbeat watchdog's fatal-escalation path directly.
    +    adapter._set_fatal_error(
    +        "telegram_network_error",
    +        "Telegram reconnect task wedged; forcing gateway reconnect.",
    +        retryable=True,
    +    )
    +    heartbeat_task = asyncio.create_task(adapter._handoff_polling_fatal_error())
    +    adapter._polling_heartbeat_task = heartbeat_task
    +    result = await asyncio.gather(heartbeat_task, return_exceptions=True)
    +
    +    assert result == [None]
    +    assert runner.adapters == {}
    +    assert Platform.TELEGRAM in runner._failed_platforms
    +
    +
     # ---------------------------------------------------------------------------
     # Connection pool drain tests (PR #16466 salvage)
     # ---------------------------------------------------------------------------