fix(telegram): bind strict cold-start readiness to its own polling generation

Follow-up hardening for the salvaged #69240 readiness gate (#67498):

- _start_polling_once now returns its (generation, progress_event) pair
  so the strict cold-start gate binds to exactly the generation it
  started, instead of re-reading self._polling_progress_event which a
  concurrent recovery task may have replaced with a newer generation's
  event (the G1/G2 race flagged in the #69240 review).
- Strict cold start no longer schedules background polling recovery: a
  polling error during the readiness wait is captured by a strict
  callback and fails the connect attempt immediately with a loud
  OSError, so GatewayRunner disposes the partial adapter and retries
  with a fresh one — no more waiting out the full readiness deadline on
  a generation that already errored, and no G2-on-partial-app healing.
- After readiness is proven the strict callback delegates every later
  polling error to the real background-recovery callback, preserving
  the existing degraded/reconnect semantics for the polling lifetime.
- The readiness-timeout error message now states the deadline and that
  the gateway will retry with a fresh adapter (loud failure, not a
  silent wait).
- Regression tests: current-generation progress connects; a polling
  error during strict cold start fails fast without scheduling
  background recovery (the #67498 idle-threads shape); stale-generation
  progress is rejected.

Progresses #67498

contributors/emails/mannnrachman@users.noreply.github.com

@@ -0,0 +1 @@
+mannnrachman

plugins/platforms/telegram/adapter.py

@@ -155,6 +155,16 @@ async def _await_with_thread_deadline(awaitable, timeout: float, *, on_abandon=N
         loop_processed_expiry.set()
 
 
+async def _first_completed(*futures: "asyncio.Future") -> None:
+    """Return when the first of ``futures`` completes.
+
+    Used by the strict cold-start readiness gate to wait on "progress OR
+    polling error", whichever fires first (#67498). Does not cancel the
+    losers — the caller owns their lifecycle.
+    """
+    await asyncio.wait(set(futures), return_when=asyncio.FIRST_COMPLETED)
+
+
 async def _run_abandon_cleanup(on_abandon) -> None:
     """Run the abandonment cleanup coroutine, swallowing any failure.
 
@@ -2134,8 +2144,16 @@ class TelegramAdapter(BasePlatformAdapter):
         drop_pending_updates: bool,
         error_callback,
         abandon_app_on_timeout: bool = False,
-    ) -> None:
-        """Start one generation and verify real getUpdates progress."""
+        schedule_verifier: bool = True,
+    ) -> tuple[int, asyncio.Event]:
+        """Start one generation and verify real getUpdates progress.
+
+        Returns the ``(generation, progress_event)`` pair created for this
+        polling generation so callers that must gate on readiness (strict
+        cold start, #67498) can bind to exactly this generation instead of
+        re-reading ``self._polling_progress_event`` — which a concurrent
+        recovery task may have replaced with a newer generation's event.
+        """
         if getattr(self, "_polling_teardown_started", False):
             raise _PollingLifecycleAbort("Telegram polling teardown started")
         generation, progress = self._begin_polling_generation()
@@ -2179,7 +2197,9 @@ class TelegramAdapter(BasePlatformAdapter):
             self._polling_progress_accepting = False
             self._send_path_degraded = True
             raise _PollingLifecycleAbort("Telegram polling teardown started")
-        self._schedule_polling_progress_verifier(generation, progress)
+        if schedule_verifier:
+            self._schedule_polling_progress_verifier(generation, progress)
+        return generation, progress
 
     def _schedule_polling_progress_verifier(
         self, generation: int, progress: asyncio.Event
@@ -2333,6 +2353,39 @@ class TelegramAdapter(BasePlatformAdapter):
             return False
         if not (self._app and self._app.updater):
             raise RuntimeError("Telegram application/updater not initialized")
+
+        # Strict cold start (#67498): background recovery must not run while
+        # the readiness gate is waiting. A G1 polling error would otherwise
+        # schedule _handle_polling_network_error(), which starts generation
+        # G2 on the same partial application while this coroutine still waits
+        # on G1's event — the cold connect then either times out on G1 despite
+        # G2 succeeding, or G2 "heals" the partial app so GatewayRunner never
+        # disposes it and retries fresh. Instead, capture the first polling
+        # error and fail the cold attempt immediately; GatewayRunner owns
+        # disposal and retry with a fresh adapter.
+        strict_error: list[BaseException] = []
+        strict_error_event = asyncio.Event()
+        strict_gate_open = True
+        effective_callback = error_callback
+        if require_progress:
+            loop = asyncio.get_running_loop()
+
+            def _strict_error_callback(error: Exception) -> None:
+                # PTB registers this callback for the whole polling
+                # generation. After the readiness gate closes (success),
+                # delegate to the real callback so ongoing polling errors
+                # keep flowing into background recovery.
+                if not strict_gate_open:
+                    if error_callback is not None:
+                        error_callback(error)
+                    return
+                if not strict_error:
+                    strict_error.append(error)
+                # PTB invokes error callbacks from the polling task; the
+                # event must be set on the loop to wake the strict waiter.
+                loop.call_soon_threadsafe(strict_error_event.set)
+
+            effective_callback = _strict_error_callback
         try:
             # Same watchdog bound as the reconnect ladders: a wedged httpx
             # connection pool can hang start_polling() forever at bootstrap
@@ -2340,26 +2393,55 @@ class TelegramAdapter(BasePlatformAdapter):
             # TimeoutError (OSError subclass), so the except below classifies
             # it via _looks_like_network_error and schedules background
             # recovery instead of blocking connect() indefinitely.
-            await self._start_polling_once(
+            generation, progress = await self._start_polling_once(
                 self._app,
                 drop_pending_updates=drop_pending_updates,
-                error_callback=error_callback,
+                error_callback=effective_callback,
                 abandon_app_on_timeout=require_progress,
+                # The strict gate below IS the cold-start verifier; the
+                # background verifier would only race it on the partial app.
+                schedule_verifier=not require_progress,
             )
             if require_progress:
+                # Bind to THIS generation's progress event (returned above),
+                # not self._polling_progress_event — a concurrent task could
+                # have replaced it with a later generation's event.
+                progress_wait = asyncio.ensure_future(progress.wait())
+                error_wait = asyncio.ensure_future(strict_error_event.wait())
                 try:
                     await _await_with_thread_deadline(
-                        self._polling_progress_event.wait(),
+                        _first_completed(progress_wait, error_wait),
                         timeout=_INITIAL_POLLING_PROGRESS_TIMEOUT,
                     )
                 except asyncio.TimeoutError as exc:
                     raise OSError(
-                        "Telegram getUpdates made no progress during initial connect"
+                        "Telegram getUpdates made no progress within "
+                        f"{_INITIAL_POLLING_PROGRESS_TIMEOUT:.0f}s during initial "
+                        "connect — failing startup so the gateway retries with a "
+                        "fresh adapter (#67498)"
                     ) from exc
-                if not self._polling_progress_event.is_set():
+                finally:
+                    for fut in (progress_wait, error_wait):
+                        if not fut.done():
+                            fut.cancel()
+                    await asyncio.gather(
+                        progress_wait, error_wait, return_exceptions=True
+                    )
+                if strict_error and not progress.is_set():
+                    raise OSError(
+                        "Telegram polling errored before first getUpdates "
+                        "success during initial connect: "
+                        f"{_redact_telegram_error_text(strict_error[0])}"
+                    ) from strict_error[0]
+                if not progress.is_set():
                     raise OSError(
                         "Telegram getUpdates did not become ready during initial connect"
                     )
+                # Readiness proven — close the strict gate so any later
+                # polling error flows to the real background-recovery
+                # callback instead of the (now finished) cold-start gate.
+                strict_gate_open = False
+                self._polling_error_callback_ref = error_callback
             return True
         except _PollingLifecycleAbort:
             return False

tests/gateway/test_telegram_start_polling_timeout.py

@@ -170,3 +170,110 @@ async def test_initial_connect_requires_get_updates_progress(monkeypatch):
             error_callback=None,
             require_progress=True,
         )
+
+
+@pytest.mark.asyncio
+async def test_initial_connect_succeeds_on_current_generation_progress(monkeypatch):
+    """Strict cold start returns True once THIS generation records progress."""
+    monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 5.0)
+    a = _bare_adapter()
+    app = MagicMock()
+    app.updater = AsyncMock()
+
+    async def start_polling_with_progress(**_kwargs):
+        # PTB's instrumented getUpdates request records progress for the
+        # generation started by this call.
+        a._record_polling_progress(a._polling_generation)
+
+    app.updater.start_polling = AsyncMock(side_effect=start_polling_with_progress)
+    a._app = app
+
+    ok = await asyncio.wait_for(
+        a._start_polling_resilient(
+            drop_pending_updates=True,
+            error_callback=None,
+            require_progress=True,
+        ),
+        timeout=10,
+    )
+    assert ok is True
+    assert a._send_path_degraded is False
+    # Strict mode owns readiness itself; the background verifier must not be
+    # racing it on the cold-start application.
+    assert getattr(a, "_polling_progress_verifier_task", None) is None
+
+
+@pytest.mark.asyncio
+async def test_initial_connect_polling_error_fails_fast_not_background(monkeypatch):
+    """#67498 idle-threads shape: a polling error during strict cold start
+    must surface immediately as a connect failure — NOT be swallowed into a
+    background recovery task that restarts polling on the partial app while
+    the readiness gate waits out its full deadline (the G1/G2 race from the
+    #69240 review)."""
+    monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 30.0)
+    a = _bare_adapter()
+    app = MagicMock()
+    app.updater = AsyncMock()
+
+    captured_callbacks = []
+
+    async def start_polling_capture(**kwargs):
+        captured_callbacks.append(kwargs.get("error_callback"))
+
+    app.updater.start_polling = AsyncMock(side_effect=start_polling_capture)
+    a._app = app
+
+    recovery_scheduled = []
+    a._schedule_polling_recovery = lambda err, reason: recovery_scheduled.append(err)
+
+    async def run_connect():
+        return await a._start_polling_resilient(
+            drop_pending_updates=True,
+            error_callback=lambda e: recovery_scheduled.append(e),
+            require_progress=True,
+        )
+
+    task = asyncio.ensure_future(run_connect())
+    # Let start_polling run and the strict gate begin waiting.
+    for _ in range(10):
+        await asyncio.sleep(0)
+        if captured_callbacks:
+            break
+    assert captured_callbacks and captured_callbacks[0] is not None
+
+    # PTB reports a network error on the first getUpdates (G1 error).
+    captured_callbacks[0](OSError("getUpdates failed"))
+
+    # The cold attempt must fail promptly (well before the 30s readiness
+    # deadline) with a loud OSError, and must NOT have scheduled background
+    # recovery (which would start G2 on the same partial application).
+    with pytest.raises(OSError, match="errored before first getUpdates"):
+        await asyncio.wait_for(task, timeout=5)
+    assert recovery_scheduled == []
+
+
+@pytest.mark.asyncio
+async def test_initial_connect_ignores_stale_generation_progress(monkeypatch):
+    """Progress recorded for a stale generation must not satisfy readiness."""
+    monkeypatch.setattr(tg_adapter, "_INITIAL_POLLING_PROGRESS_TIMEOUT", 0.1)
+    a = _bare_adapter()
+    app = MagicMock()
+    app.updater = AsyncMock()
+
+    async def start_polling_stale_progress(**_kwargs):
+        # Progress arrives tagged with a PREVIOUS generation (e.g. a late
+        # response from an abandoned attempt) — must be rejected.
+        a._record_polling_progress(a._polling_generation - 1)
+
+    app.updater.start_polling = AsyncMock(side_effect=start_polling_stale_progress)
+    a._app = app
+
+    with pytest.raises(OSError, match="getUpdates made no progress"):
+        await asyncio.wait_for(
+            a._start_polling_resilient(
+                drop_pending_updates=True,
+                error_callback=None,
+                require_progress=True,
+            ),
+            timeout=10,
+        )