fix(telegram): keep update receipts across adapter rebuilds and restarts (#120257)

Completed update IDs lived only in the adapter's memory. The gateway
reconnect watcher builds a new TelegramAdapter and connects it with
is_reconnect=True, which keeps Telegram's pending queue, and a new PTB
Updater polls from offset 0. Telegram then resends every update whose
acknowledgement (the next getUpdates offset, or the cleanup call in
Updater.stop) never landed, and the fresh adapter admitted them again.

Write completed IDs to telegram_update_receipts_<bot_id>.json in the
adapter's Hermes home and seed admission from it once per bot. Receipts
older than 24h are dropped: the Bot API keeps unconfirmed updates no
longer than that, and it keeps the lookup clear of the random ID restart
Telegram may do after a week without updates. Writes are coalesced and
run off the loop; disconnect waits for the last one.

Refs #68502

Co-authored-by: Joe Githler <5716896+NoTimeforInfinity@users.noreply.github.com>

plugins/platforms/telegram/adapter.py

@@ -544,6 +544,13 @@ class TelegramAdapter(BasePlatformAdapter):
         self._seen_update_ids: dict = {}
         self._inflight_update_ids: dict = {}
         self._update_admission = None
+        # Completed update IDs survive adapter replacement and restarts (update_admission.py).
+        # Resolved now: secondary profiles construct adapters inside their own home scope.
+        from hermes_constants import get_hermes_home
+        self._update_receipt_dir = get_hermes_home()
+        self._update_receipts_loaded: set = set()
+        self._update_receipts_dirty: set = set()
+        self._update_receipt_flush: Optional[asyncio.Task] = None
         self._bot: Optional[Bot] = None
         self._webhook_mode: bool = False
         self._mention_patterns = self._compile_mention_patterns()
@@ -3453,6 +3460,10 @@ class TelegramAdapter(BasePlatformAdapter):
                 logger.warning("[%s] Error during Telegram disconnect: %s", self.name, _redact_telegram_error_text(e))
         self._app = None
         self._bot = None
+        # Land the last completed receipts before a replacement adapter reads them.
+        flush = getattr(self, "_update_receipt_flush", None)
+        if flush is not None and not flush.done():
+            await self._await_disconnect_step(asyncio.shield(flush), _DISCONNECT_STEP_TIMEOUT, "update-receipt flush")
         logger.info("[%s] Disconnected from Telegram", self.name)
 
     def _should_thread_reply(self, reply_to: Optional[str], chunk_index: int) -> bool:

plugins/platforms/telegram/update_admission.py

@@ -1,11 +1,17 @@
 """Telegram admission before every PTB handler group, including native plugins.
 
 Claims belong to the receiving adapter across Application rebuilds. Completed
-history is bounded and has no TTL; dispatch and its PTB tasks pin active claims.
-No disk receipt, cross-process coordination or exactly-once effects are promised.
+history is bounded and has no TTL in memory; dispatch and its PTB tasks pin active
+claims. Completed IDs are also written to a per-bot receipt file under the adapter's
+Hermes home, so a replacement adapter or a restarted gateway still drops updates that
+Telegram redelivers because their getUpdates/webhook acknowledgement never landed.
+No cross-process coordination or exactly-once effects are promised.
 """
 
 import asyncio
+import json
+import logging
+import time
 from contextvars import ContextVar
 from dataclasses import dataclass
 from functools import wraps
@@ -14,9 +20,74 @@ from telegram import Update
 from telegram.ext import Application, ApplicationHandlerStop, ConversationHandler
 
 from gateway.platforms.helpers import bounded_put
+from utils import atomic_json_write
 
+logger = logging.getLogger(__name__)
 
 _DEFAULT_BLOCK = object()
+_SEEN_CAP = 4096
+# The Bot API keeps an unconfirmed update for at most 24 hours (getUpdates), so an older
+# receipt can never match a redelivery. That is also well inside the week of silence after
+# which Telegram may restart update IDs at a random value, so a set lookup stays safe
+# where a persisted numeric high-watermark would not.
+RECEIPT_TTL_SECONDS = 24 * 60 * 60
+
+
+def _receipt_path(adapter, bot_id):
+    return adapter._update_receipt_dir / f"telegram_update_receipts_{bot_id}.json"
+
+
+def _load_receipts(adapter, bot_id) -> None:
+    """Seed completed history from disk once per bot, before its first admission check."""
+    if bot_id in adapter._update_receipts_loaded:
+        return
+    adapter._update_receipts_loaded.add(bot_id)
+    path = _receipt_path(adapter, bot_id)
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return
+    except (OSError, ValueError):
+        logger.warning("[Telegram] Ignoring unreadable update receipts at %s", path, exc_info=True)
+        return
+    ids = payload.get("update_ids") if isinstance(payload, dict) else None
+    if not isinstance(ids, dict):
+        return
+    cutoff = time.time() - RECEIPT_TTL_SECONDS
+    fresh = sorted((float(ts), uid) for uid, ts in ids.items()
+                   if isinstance(ts, (int, float)) and ts > cutoff and str(uid).lstrip("-").isdigit())
+    seen = adapter._seen_update_ids
+    for ts, uid in fresh:
+        key = f"{bot_id}:{uid}"
+        if key not in seen:
+            bounded_put(seen, key, ts, _SEEN_CAP)
+
+
+def _record_receipt(adapter, key: str) -> None:
+    bounded_put(adapter._seen_update_ids, key, time.time(), _SEEN_CAP)
+    adapter._update_receipts_dirty.add(key.split(":", 1)[0])
+    task = adapter._update_receipt_flush
+    if task is None or task.done():
+        # One writer per adapter coalesces bursts; the fsync stays off the event loop.
+        adapter._update_receipt_flush = asyncio.get_running_loop().create_task(
+            _flush_receipts(adapter), name="telegram-update-receipts")
+
+
+async def _flush_receipts(adapter) -> None:
+    from hermes_constants import mkdir_under_hermes_home
+
+    while adapter._update_receipts_dirty:
+        bot_id = adapter._update_receipts_dirty.pop()
+        prefix, cutoff = f"{bot_id}:", time.time() - RECEIPT_TTL_SECONDS
+        # Snapshot on the loop: admission mutates this dict only from loop callbacks.
+        ids = {key[len(prefix):]: ts for key, ts in adapter._seen_update_ids.items()
+               if key.startswith(prefix) and isinstance(ts, float) and ts > cutoff}
+        path = _receipt_path(adapter, bot_id)
+        try:
+            mkdir_under_hermes_home(path.parent)
+            await asyncio.to_thread(atomic_json_write, path, {"update_ids": ids}, indent=None)
+        except Exception:
+            logger.warning("[Telegram] Failed to persist update receipts to %s", path, exc_info=True)
 
 
 @dataclass
@@ -140,7 +211,7 @@ class TelegramApplication(Application):
             return
         del self.adapter._inflight_update_ids[claim.key]
         if claim.accepted or (claim.completed and not claim.failed):
-            bounded_put(self.adapter._seen_update_ids, claim.key, None, 4096)
+            _record_receipt(self.adapter, claim.key)
 
     async def process_error(self, update, error, job=None, coroutine=None):
         claim = self._current_claim.get()
@@ -153,7 +224,9 @@ class TelegramApplication(Application):
     async def process_update(self, update):
         if not isinstance(update, Update):
             return await super().process_update(update)
-        key = f"{self.bot.id}:{update.update_id}"
+        bot_id = self.bot.id
+        key = f"{bot_id}:{update.update_id}"
+        _load_receipts(self.adapter, bot_id)
         # Dispatch happened even when preparation fails before the group-99 observer.
         self.adapter._updates_dispatched_total += 1
         seen = self.adapter._seen_update_ids

tests/plugins/test_telegram_update_admission.py

@@ -84,7 +84,7 @@ def update(bot, uid=10, kind="text", *, edited=False, chat=42, text="hello", gro
 
 
 @asynccontextmanager
-async def connected(monkeypatch, *, extra=None, bot_id=111):
+async def connected(monkeypatch, *, extra=None, bot_id=111, is_reconnect=False):
     adapter = TelegramAdapter(PlatformConfig(enabled=True, token=f"{bot_id}:offline-test", extra=extra or {}))
     # Only transport/lifecycle services and the final model-work boundary are replaced.
     monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "88")
@@ -102,7 +102,7 @@ async def connected(monkeypatch, *, extra=None, bot_id=111):
         return True
 
     monkeypatch.setattr(adapter, "_start_session_processing", start)
-    assert await adapter.connect()
+    assert await adapter.connect(is_reconnect=is_reconnect)
     try:
         yield adapter, adapter._app, delivered
     finally:
@@ -192,13 +192,14 @@ async def test_replay_is_admitted_once_before_dispatch(monkeypatch, tmp_path, ki
             await app.process_update(update(app.bot, **args))
             assert len(delivered) == 2
         if mode == "owners":
-            for profile, bot_id in (("alpha", 222), ("beta", 333), ("alpha", 222)):
+            for profile, bot_id, fresh in (("alpha", 222, 1), ("beta", 333, 1), ("alpha", 222, 0)):
                 token = set_hermes_home_override(tmp_path / profile)
                 try:
                     async with connected(monkeypatch, bot_id=bot_id) as (other, other_app, other_delivered):
                         await other_app.process_update(update(other_app.bot))
                         await asyncio.gather(*other._pending_text_batch_tasks.values())
-                        assert len(other_delivered) == 1
+                        # A rebuilt adapter for the same bot and home reads that home's receipt.
+                        assert len(other_delivered) == fresh
                 finally:
                     reset_hermes_home_override(token)
             await app.process_update(update(app.bot))
@@ -739,3 +740,53 @@ async def test_only_pre_handoff_failure_reopens_admission(monkeypatch, tmp_path,
         assert len(delivered) == 1
         assert delivered[0].text == ("/status" if stage == "dispatch" else "hello")
         assert adapter._platform_event_handler.await_count == (1 if stage in ("prepare", "pressure", "batch_prepare") else 0)
+
+
+@pytest.mark.asyncio
+async def test_redelivery_to_rebuilt_adapter_is_dropped(monkeypatch, tmp_path):
+    """The reconnect watcher and a gateway restart both build a new adapter, and a new PTB
+    Updater polls from offset 0: Telegram resends every update whose acknowledgement never
+    landed. The receipt must outlive the adapter that completed the update."""
+    from hermes_constants import get_hermes_home
+    from plugins.platforms.telegram.update_admission import RECEIPT_TTL_SECONDS
+
+    receipts = get_hermes_home() / "telegram_update_receipts_111.json"
+    async with connected(monkeypatch) as (adapter, app, delivered):
+        await app.process_update(update(app.bot, 10))
+        with monkeypatch.context() as broken:
+            broken.setattr(adapter, "_cache_replied_media", AsyncMock(side_effect=OSError("before enqueue")))
+            await app.process_update(update(app.bot, 20, text="unaccepted"))
+        await asyncio.gather(*adapter._pending_text_batch_tasks.values())
+        assert [event.text for event in delivered] == ["hello"]
+    # disconnect() waits for the receipt write; the failed preparation is not a receipt.
+    assert set(json.loads(receipts.read_text())["update_ids"]) == {"10"}
+
+    # A fresh adapter, connected the way the gateway reconnect watcher does it.
+    async with connected(monkeypatch, is_reconnect=True) as (adapter, app, delivered):
+        await app.process_update(update(app.bot, 10))
+        await app.process_update(update(app.bot, 20, text="unaccepted"))
+        await asyncio.gather(*adapter._pending_text_batch_tasks.values())
+        await app.process_update(update(app.bot, 21, edited=True, text="changed"))
+        await asyncio.gather(*adapter._pending_text_batch_tasks.values())
+        # Old update dropped; the retry of an unaccepted one and an edit of message 472 still run.
+        assert [event.text for event in delivered] == ["unaccepted", "changed"]
+        assert adapter._updates_dispatched_total == 3
+
+    # Receipts are bot-scoped, and older than Telegram's 24h retention they cannot match.
+    async with connected(monkeypatch, bot_id=222) as (adapter, app, delivered):
+        await app.process_update(update(app.bot, 10))
+        await asyncio.gather(*adapter._pending_text_batch_tasks.values())
+        assert len(delivered) == 1
+    stale = time.time() - RECEIPT_TTL_SECONDS - 1
+    receipts.write_text(json.dumps({"update_ids": {"10": stale, "20": "bad", "x": time.time()}}))
+    async with connected(monkeypatch) as (adapter, app, delivered):
+        await app.process_update(update(app.bot, 10))
+        await asyncio.gather(*adapter._pending_text_batch_tasks.values())
+        assert len(delivered) == 1
+    assert set(json.loads(receipts.read_text())["update_ids"]) == {"10"}
+
+    receipts.write_text("{not json")
+    async with connected(monkeypatch) as (adapter, app, delivered):
+        await app.process_update(update(app.bot, 30))
+        await asyncio.gather(*adapter._pending_text_batch_tasks.values())
+        assert len(delivered) == 1