fix(gateway): hand Telegram held inbound to the rebuilt adapter

Telegram holds inbound events that PTB has already acked (offset advanced),
so dropping them means losing the message for good. The hold queue lives on
the adapter instance and drains only on that instance's _mark_connected. When
a retryable fatal (network error, polling/ingress stall) makes the runner
build a NEW adapter, that drain never runs: the reconnect path carried the
inbound dedup caches across but not the queue. Held events, plus anything
teardown salvaged onto the retired instance later, were silently stranded.

The reconnect queue entry (primary) and the secondary reconnect task now
keep the replaced adapter as "predecessor". The existing carry_inbound_dedup
seam reads its dedup caches from it before connect. A new
hand_over_held_inbound seam moves its held events at publish time
(_install_reconnected_adapter / secondary profile_map install). A candidate
that fails to connect never owns the queue, so no hand-back is needed, and
delivery waits until the replacement is registered. TelegramAdapter.
adopt_held_inbound rebinds only source._transport_adapter_ref (profile and
role authorization stay as resolved) and leaves a successor weakref on the
predecessor. Late _hold_inbound_event calls there (teardown salvage after the
replacement drained) forward to the replacement.

Reported-by: j2h4u (Maxim Brashenko)
Fixes #132829

gateway/platforms/helpers.py

@@ -72,18 +72,28 @@ def inbound_dedup_caches(adapter: Any) -> dict[str, MessageDeduplicator]:
     return {name: v for name, v in vars(adapter).items() if isinstance(v, MessageDeduplicator)}
 
 
-def carry_inbound_dedup(caches: Optional[dict], adapter: Any) -> None:
-    """Seed a rebuilt adapter's dedup caches from the instance it replaces.
+def carry_inbound_dedup(predecessor: Any, adapter: Any) -> None:
+    """Seed a rebuilt adapter's dedup caches from the instance it replaces (call before connect).
 
     The runner's reconnect path builds a NEW adapter; without this a platform replaying a recent
     inbound ID after the reconnect (websocket resume, webhook retry, unacked poll batch) is
     admitted and answered a second time."""
-    for name, previous in (caches or {}).items():
+    for name, previous in (inbound_dedup_caches(predecessor) if predecessor is not None else {}).items():
         current = getattr(adapter, name, None)
         if isinstance(current, MessageDeduplicator) and current is not previous:
             current.absorb(previous)
 
 
+def hand_over_held_inbound(predecessor: Any, adapter: Any) -> None:
+    """Move inbound the retired instance is holding to its now-published replacement (#132829).
+
+    Runs at publish time, not before connect: a candidate that fails to connect never owns the
+    queue, and the predecessor keeps holding until a replacement is registered."""
+    adopt = getattr(adapter, "adopt_held_inbound", None)
+    if predecessor is not None and callable(adopt):
+        adopt(predecessor)
+
+
 # Worker-thread handoff used by the off-loop persist paths.  A module attribute
 # so tests can replace THIS seam instead of patching ``asyncio.to_thread``
 # globally.

gateway/run_adapters.py

@@ -26,7 +26,7 @@ from gateway.config import (
     platform_binds_port as _platform_binds_port,
 )
 from gateway.platforms.base import BasePlatformAdapter
-from gateway.platforms.helpers import carry_inbound_dedup, inbound_dedup_caches
+from gateway.platforms.helpers import carry_inbound_dedup, hand_over_held_inbound
 from gateway.restart import is_global_startup_conflict
 from gateway.run_shutdown import _log_suppressed
 from gateway.session import SessionSource
@@ -244,7 +244,8 @@ class GatewayAdapterLifecycleMixin:
             **({"queued_at": now} if queued else {}),
             "credential_claim": self._adapter_credential_claim(platform, adapter),
             "listener_claim": self._adapter_listener_claim(platform, adapter),
-            "inbound_dedup": inbound_dedup_caches(adapter) if adapter is not None else None,
+            # The replaced instance: its dedup caches seed each candidate, its held inbound moves on publish.
+            "predecessor": adapter,
         }
 
     def _queue_retryable_fatal_platform(self, adapter: BasePlatformAdapter) -> bool:
@@ -827,7 +828,7 @@ class GatewayAdapterLifecycleMixin:
                 )
                 logger.info("Reconnect %s: no adapter yet, next retry in %ds", platform.value, backoff)
                 return
-            carry_inbound_dedup(info.get("inbound_dedup"), adapter)
+            carry_inbound_dedup(info.get("predecessor"), adapter)
             self._wire_adapter_handlers(adapter)
             # is_reconnect keeps the server-side update queue so offline-period messages are delivered.
             success = await self._connect_adapter_with_timeout(adapter, platform, is_reconnect=True)
@@ -891,7 +892,7 @@ class GatewayAdapterLifecycleMixin:
         """Publish a freshly reconnected primary adapter and replay what it missed while down."""
         self._publish_primary_adapter(platform, adapter)
         self.delivery_router.adapters = self.adapters
-        del self._failed_platforms[platform]
+        hand_over_held_inbound(self._failed_platforms.pop(platform).get("predecessor"), adapter)
         # connect() returning True does not mean the receive path is confirmed -- Telegram's degraded
         # reconnect returns True so the gateway stays up while its own ladder retries. Stamping "connected"
         # here would undo the adapter's accurate status.
@@ -1398,7 +1399,7 @@ class GatewayAdapterLifecycleMixin:
                 and _platform_binds_port(platform.value, getattr(getattr(adapter, "config", None), "extra", None)):
             adapter._shared_listener_profile = profile_name
 
-    async def _secondary_reconnect_attempt(self, profile_name: str, platform: Platform, inbound_dedup=None):
+    async def _secondary_reconnect_attempt(self, profile_name: str, platform: Platform, predecessor=None):
         """One scoped attempt to rebuild+connect a secondary adapter → ``(adapter, success)``;
         ``(None, None)`` = give up for good (disabled, credential removed, adapter unavailable). Caller
         tears down a RETURNED adapter; one whose configure/connect raised is torn down here."""
@@ -1430,7 +1431,7 @@ class GatewayAdapterLifecycleMixin:
                     platform.value, profile_name,
                 )
                 return None, None
-            carry_inbound_dedup(inbound_dedup, adapter)
+            carry_inbound_dedup(predecessor, adapter)
             try:
                 self._configure_profile_adapter(adapter, profile_name, platform)
                 success = await self._connect_adapter_with_timeout(adapter, platform, is_reconnect=True)
@@ -1441,7 +1442,7 @@ class GatewayAdapterLifecycleMixin:
             return adapter, success
 
     async def _run_secondary_profile_reconnect(
-        self, profile_name: str, platform: Platform, inbound_dedup=None
+        self, profile_name: str, platform: Platform, predecessor=None
     ) -> None:
         """Reconnect a retryable secondary adapter under its own profile scope."""
         from gateway.run import _profile_runtime_scope, _reconnect_backoff
@@ -1454,7 +1455,7 @@ class GatewayAdapterLifecycleMixin:
                 adapter = None
                 try:
                     adapter, success = await self._secondary_reconnect_attempt(
-                        profile_name, platform, inbound_dedup
+                        profile_name, platform, predecessor
                     )
                     if adapter is None:
                         return
@@ -1462,6 +1463,7 @@ class GatewayAdapterLifecycleMixin:
                         profile_map = self._profile_adapters.setdefault(profile_name, {})
                         if platform not in profile_map:
                             profile_map[platform] = adapter
+                            hand_over_held_inbound(predecessor, adapter)
                             self._sync_voice_mode_state_to_adapter(adapter)
                             logger.info("✓ %s reconnected (profile: %s)", platform.value, profile_name)
                             await self._redeliver_failed_obligations_for_platform(
@@ -1576,7 +1578,7 @@ class GatewayAdapterLifecycleMixin:
         if platform in profile_pending:
             return
         profile_pending[platform] = self._retain_background_task(asyncio.create_task(
-            self._run_secondary_profile_reconnect(profile_name, platform, inbound_dedup_caches(adapter)),
+            self._run_secondary_profile_reconnect(profile_name, platform, adapter),
             name=f"secondary-reconnect:{profile_name}:{platform.value}",
         ))
 

plugins/platforms/telegram/adapter.py

@@ -10,6 +10,7 @@ import os
 import html as _html
 import re
 import time
+import weakref
 from contextvars import ContextVar
 from datetime import datetime, timezone
 from typing import Any, Awaitable, Callable, Dict, Iterator, List, Optional, Set
@@ -818,6 +819,10 @@ class TelegramAdapter(BasePlatformAdapter):
             logger.warning(
                 "[Telegram] Discarding inbound under non-retryable fatal (%s, %d chars)", where, len(getattr(event, "text", None) or ""))
             return
+        successor = getattr(self, "_held_inbound_successor", None)
+        if successor is not None and successor() is not None:  # retired by a runner rebuild (#132829)
+            successor()._adopt_held_event(event, where=f"{where}-forwarded", schedule=schedule)
+            return
         held = getattr(self, "_held_inbound_events", None)
         if held is None:
             self._held_inbound_events = held = []
@@ -837,6 +842,24 @@ class TelegramAdapter(BasePlatformAdapter):
         if schedule and not self._should_drop_delayed_delivery():
             self._schedule_held_inbound_redispatch()
 
+    def adopt_held_inbound(self, predecessor: "TelegramAdapter") -> None:
+        """Take over the hold queue of the instance the runner just replaced with us (#132829): it only
+        drains on its own ``_mark_connected``, which never comes; later holds there forward here."""
+        if predecessor is self or not isinstance(predecessor, TelegramAdapter):
+            return
+        predecessor._held_inbound_successor = weakref.ref(self)
+        held = getattr(predecessor, "_held_inbound_events", None) or []
+        events, held[:] = list(held), []
+        for event in events:
+            self._adopt_held_event(event, where="adopted", schedule=False)
+        self._schedule_held_inbound_redispatch()
+
+    def _adopt_held_event(self, event: "MessageEvent", *, where: str, schedule: bool) -> None:
+        """Hold a predecessor's event here; only the transport ref moves (profile/authz stay as resolved)."""
+        if getattr(event, "source", None) is not None:
+            event.source._transport_adapter_ref = weakref.ref(self)
+        self._hold_inbound_event(event, where=where, schedule=schedule)
+
     def _rehold_from(self, events: list, idx: int, where: str) -> None:
         """Re-hold ``events[idx:]`` without rescheduling (drain interrupted / failed / cancelled)."""
         for rest in events[idx:]:

tests/gateway/test_platform_reconnect.py

@@ -412,6 +412,49 @@ class TestReconnectKeepsInboundDedup:
         assert new._dedup.is_duplicate("m1") is True
         assert new._dedup.is_duplicate("m2") is False
 
+    @pytest.mark.asyncio
+    async def test_held_telegram_inbound_reaches_published_replacement_once(self):
+        """PTB already acked a held update, so the queue on the retired Telegram adapter must be
+        delivered exactly once by the replacement the watcher publishes, not by a failed candidate (#132829)."""
+        from gateway.platforms.event import MessageEvent
+        from gateway.session import SessionSource
+        from plugins.platforms.telegram.adapter import TelegramAdapter
+
+        class _Telegram(TelegramAdapter):
+            def __init__(self, succeed):
+                super().__init__(PlatformConfig(enabled=True, token="123:abc"))
+                self.succeed, self.handle_message = succeed, AsyncMock()
+
+            async def connect(self, *, is_reconnect=False):
+                if self.succeed:
+                    self._mark_connected()
+                return self.succeed
+
+            async def disconnect(self):
+                return None
+
+        runner = _make_runner()
+        runner.stop = AsyncMock()
+        runner._sync_voice_mode_state_to_adapter = MagicMock()
+        old, failed, new = _Telegram(True), _Telegram(False), _Telegram(True)
+        runner.adapters[Platform.TELEGRAM] = old
+        old._set_fatal_error("telegram_network_error", "stall", retryable=True)
+        event = MessageEvent(text="held", source=SessionSource(platform=Platform.TELEGRAM, chat_id="1", chat_type="dm"))
+        old._hold_inbound_event(event, where="text-enqueue")
+        await runner._handle_adapter_fatal_error(old)
+        for candidate in (failed, new):
+            runner._failed_platforms[Platform.TELEGRAM]["next_retry"] = 0
+            with patch.object(runner, "_create_adapter", return_value=candidate):
+                await runner._reconnect_failed_platform(Platform.TELEGRAM, time.monotonic() + 1)
+        await asyncio.sleep(0)
+        await new._held_inbound_redispatch_task
+
+        assert runner.adapters[Platform.TELEGRAM] is new
+        failed.handle_message.assert_not_called()
+        new.handle_message.assert_awaited_once_with(event)
+        assert event.source._transport_adapter_ref() is new
+        assert old._held_inbound_events == [] and new._held_inbound_events == []
+
 
 # --- Pause / resume circuit breaker ---
 

tests/gateway/test_telegram_text_batching.py

@@ -195,6 +195,21 @@ class TestHoldInboundAcrossReconnect:
         assert adapter.handle_message.call_args[0][0].text == "should survive disconnect"
         assert adapter._held_inbound_events == []
 
+    @pytest.mark.asyncio
+    async def test_late_teardown_salvage_on_retired_adapter_reaches_replacement(self):
+        """Teardown of a rebuilt-away adapter can salvage a batch after the replacement drained (#132829)."""
+        old, new = _make_adapter(), _make_adapter()
+        old._mark_disconnected()
+        new.adopt_held_inbound(old)
+        old._pending_text_batches["k"] = _make_event("late-salvage")
+
+        await old._cancel_pending_delivery_tasks()
+        await new._held_inbound_redispatch_task
+
+        old.handle_message.assert_not_called()
+        assert [c.args[0].text for c in new.handle_message.call_args_list] == ["late-salvage"]
+        assert old._held_inbound_events == []
+
     @pytest.mark.asyncio
     async def test_flush_during_disconnect_holds_popped_event(self):
         """After pop, drop-guard must hold — not destroy — the event.