diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b3c98c7..b0f50f35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -303,6 +303,11 @@ Ports upstream `32687038` (vercel/chat#830, chat@4.38.1), the Google Chat part o - New exports from `chat_sdk.ai`: `ChatApprovalToolName`, `ReadScope`; `ToolOptions.guard`; `ChatToolsOptions.scope` / `strict_scope`. - **Python-specific (divergence from upstream):** link-metadata bounds count code points, not UTF-16 code units, so a field with emoji keeps up to the limit in code points and never ends on a lone surrogate. See `docs/UPSTREAM_SYNC.md`. - Fidelity: `ai/index.test.ts` 26 → 0 missing, `ai/messages.test.ts` 10 → 9 at `chat@4.41.1`. +- **Telegram polling is at-least-once, and albums arrive as one message** (#227; ports vercel/chat `629e6555` #760, chat@4.37.0, and the Telegram half of `91683e52` #942, chat@4.41.0). Outbound multi-file `sendMediaGroup` (#605) is split out to #278. + - **Consumer-visible (polling):** the poller now waits for every handler of a batch before advancing `offset`. A handler failure no longer loses the update: it is saved in a `telegram:polling:{sha256(bot_user_id)}` state checkpoint and retried with backoff (`max(retry_delay_ms, 1000) * 2^(attempts-1)`, capped at 30 s, at least `retry_after` on rate limits), also after a restart. Handlers must tolerate redelivery: a handler that failed (or finished just before a crash) can run again. Polled updates bypass core dedupe (`WebhookOptions(deduplicate=False)`); the checkpoint deduplicates them. A batch is handled before the next `getUpdates`, so one slow handler delays the next poll. A failed startup `getMe` is retried on every poll. + - **Consumer-visible (albums):** the parts of an album (`media_group_id`) are buffered in state for 1 s after the newest part and reach handlers as **one** `Message` (newest part's id and raw, first non-empty text, all attachments in order, `is_mention` if any part mentions the bot), on the webhook and polling paths. Handlers that saw N messages per album now see one, about 1 s later (about 2 s when polling). An album caption starting with `/command` is the album's text, not a slash command. + - `TelegramAdapter.process_update` now returns the list of dispatched handler tasks; `handle_incoming_message_update`, `handle_callback_query`, `handle_message_reaction_update` and `handle_slash_command_update` return their task(s). `TelegramMessage` gains `media_group_id`. + - `stop_polling()` interrupts only a pending `getUpdates` or polling sleep; while handlers run it waits for them, as upstream. **Python-specific:** a handler task cancelled by `Chat.shutdown` counts as a failure, so its update stays in the checkpoint, and a stop that lands while the loop reads its checkpoint skips the ready retry batch until the next start (upstream still dispatches it). See `docs/UPSTREAM_SYNC.md`. - **BREAKING (Linear agent-session thread ids) — one stable thread per agent session** (#232; ports vercel/chat `3d2cb22a` #885, chat@4.40.0, and the Linear half of `fcdc1c9e` #946, chat@4.41.0). - **Breaking:** every agent-session message (created and prompted webhooks, fetched history, and the messages returned by `post_message` / `stream`) now uses `linear:{issueId}:s:{agentSessionId}`. It used to be `linear:{issueId}:c:{commentId}:s:{agentSessionId}`, a new thread per source comment. **Migration:** subscriptions and state stored under the old ids no longer match new events. Re-subscribe, or map each stored id by dropping its `:c:{commentId}` segment. Old-form ids still decode, so posting to a stored one still reaches its session. - **Consumer-visible:** a session created without a creator (for example by a Linear automation) is authored by `Author(user_id="linear-automation", user_name="Linear automation", is_bot=True, is_me=False)` and reaches your handlers. It used to be authored as the bot itself and dropped as a self-message. diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index f9340170..35b9428f 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -1089,6 +1089,79 @@ SDK-free cards-primitives surface, so upstream's two table converters (and the modals to cards (see the "Teams dialog/modal inbound" row in the non-parity table), so only the primitive gains the new children. +### Telegram polling acknowledgement and incoming albums (chat@4.37–4.41, #227) + +Ports `629e6555` (vercel/chat#760, chat@4.37.0) and `91683e52` +(vercel/chat#942, chat@4.41.0) in `adapters/telegram/adapter.py`. The core +half of #942 (`WebhookOptions.deduplicate`, task-returning `process_*`) +landed in #191. Outbound multi-file `sendMediaGroup` (`8d7ccdb1`, #605) is +split out to #278. + +- `process_update` returns the list of dispatched handler tasks (message, + album, slash command, action, reactions). The webhook path ignores it. +- **Albums.** A part with `media_group_id` skips slash-command routing and is + buffered under `telegram:incoming-media-group:{thread}:{media_group_id}` + (state lock 5 s TTL / 50 ms retry, buffer 30 s TTL, newest 10 parts). After + 1 s with no new part, one `Message` is dispatched: id, raw, author and + metadata of the newest part (highest `message_id`), text and AST of the + first part with text, every attachment in `message_id` order, links + concatenated (`None` when no part has any), `is_mention` if any part + mentions the bot. `reply_to` is the first part's non-`None` `reply_to`, as + upstream; Telegram parsing does not populate it until #228. Typing starts once, for + the first part, after the album settles. On the webhook path the album task + is held in `_media_group_tasks` (no GC mid-settle); `wait_until` gets a + wrapper that never raises, a failure is logged as "Failed to process + incoming Telegram media group", and the task returned from `process_update` + still raises. +- **Polling.** `polling_loop` follows upstream `pollingLoop` / + `processPollingUpdates` step for step: handler tasks are awaited (per + album, `asyncio.wait` + first error rethrown, so the handler tasks are + never cancelled by the loop) before `offset` moves; failures and unsettled + album parts are kept in `telegram:polling:{sha256(bot_user_id)}` as + `{"offset", "pending": [{"update", "receivedAt", "attempts", "retryAt"}]}` + (camelCase, JSON-safe raw update dicts); the retry delay is + `max(retry_delay_ms, 1000) * 2**(attempts-1)` capped at 30 s and at least + `AdapterRateLimitError.retry_after` seconds; the loop polls once between + ready retry batches (`drained`); `getUpdates` is cut short (no updates) when + the earliest pending deadline falls due, and asks with `timeout=0` when one + is already due. Polled updates are dispatched with + `WebhookOptions(deduplicate=False)`: the checkpoint, not core dedupe, + deduplicates them, so a retry is not dropped by the 10-minute dedupe window. + `_ensure_bot_identity` runs on every iteration, so a failed startup `getMe` + recovers once Telegram is reachable. +- **Clock.** `receivedAt` / `retryAt` are epoch milliseconds + (`int(time.time() * 1000)`, upstream `Date.now()`), not monotonic, because + they are persisted and compared after a restart or on another instance. The + adapter reads time through `_now_ms()` and sleeps through `_sleep(ms)` so + tests can inject a fake clock. +- **Python-specific: stop semantics.** Upstream's per-iteration + `AbortController` reaches only `getUpdates` and the loop's sleeps. + `stop_polling` therefore cancels the polling task only while it is parked + in one of those (`_polling_abortable`); while handlers run or the + checkpoint is written it waits for that step, and the loop then exits, + as upstream's `stopPolling` does. `stop_polling` waits with + `asyncio.wait` instead of `suppress(CancelledError)`, so cancelling the + caller still propagates. One divergence: a stop that lands while the loop + awaits `_ensure_bot_identity()` or the checkpoint `state.get` makes the + loop return before dispatching the ready retry batch. Upstream has no + `pollingActive` check there and still dispatches it; in Python that batch + could start handlers after `Chat.shutdown`'s cancellation sweep. The batch + stays in the checkpoint and is retried on the next start. +- **Python-specific: cancelled handlers.** `Chat.shutdown` cancels in-flight + handler tasks before disconnecting adapters (upstream waits for them). A + cancelled handler task is counted as a failure, so the update stays in the + checkpoint and is retried after a restart instead of being acknowledged + unhandled. For the same reason `disconnect()` cancels albums still settling + (`_media_group_tasks`) before stopping the poller; otherwise an album could + dispatch a new handler after `Chat.shutdown`'s cancellation sweep and keep + `stop_polling` waiting on it. +- **Album buffer key (parity).** The buffer key is not scoped to the bot + identity, exactly as upstream. Bots that share one state namespace must not + share a group chat (core `dedupe:` keys are not bot-scoped either). +- Delivery is now at-least-once: an update whose handler finished but whose + acknowledgement never reached Telegram (a crash before the next + `getUpdates`) is delivered again, as upstream. + ### Discord correctness and security (chat@4.32–4.41, #229) Parity with upstream `adapter-discord` at chat@4.41.1 for the webhook @@ -1885,8 +1958,8 @@ stay explicit instead of being rediscovered in code review. | Teams `User-Agent: Vercel.ChatSDK` outbound header | Not set on `aiohttp` calls | Propagated by `botbuilder` 2.0.8 | Python Teams adapter doesn't use `botbuilder` (raw `aiohttp`). Upstream's vercel/chat#415 was a JS-only `botbuilder` SDK bump that flipped `X-User-Agent` → `User-Agent`. No equivalent dependency to bump on the Python side. Setting a `User-Agent` on the ~9 outbound `aiohttp` call sites would be a defense-in-depth nice-to-have; deferred to a follow-up. | | Teams adapter on `microsoft-teams-apps` (official MS Python SDK) | Inbound webhook + JWT auth, outbound send/edit/delete/typing, and native DM streaming all delegate to the official `microsoft-teams-apps` SDK `App`; Graph reads stay hand-rolled over `aiohttp` | `@microsoft/teams.apps` owns the wire format, throttling, and activity routing | **Delivered in 0.4.30** (issue #93, PRs 1–4). The migration shipped as four PRs: inbound + auth (#143), outbound (#144), native streaming via the SDK `IStreamer` (#145), and this release cut. The 3.12 floor bump (#111) — the migration's prerequisite — landed in 0.4.29. The residual adapter-level divergences (we keep the SDK as auth + transport but route the authenticated activity ourselves; close the streamer in our own `finally` because our bridge owns dispatch) are documented in the Teams divergence rows above. Graph stays hand-rolled (no `msgraph-sdk` / `[graph]` extra). | | Telegram `get_user().is_bot` | Always `False` (matches upstream — `getChat` does not expose `is_bot`) | Always `false` (same caveat documented in upstream code comment) | The Telegram Bot API's `getChat` endpoint does not surface the `is_bot` field that's available on the `User` object inside incoming `Message` updates. Callers needing bot detection must use `message.author.is_bot` from webhooks instead of `chat.get_user(...).is_bot`. | -| Telegram webhook verification + `update_id` dedupe (4.41 wave, #224) | **Parity** with upstream `c4a359e7` (vercel/chat#858, chat@4.39), `1d2b78d9` (#799, chat@4.38) and the bot-identity scope helper from `7a1150ce` (#813): webhook mode requires `secret_token` unless `allow_unverified_webhooks` / `TELEGRAM_ALLOW_UNVERIFIED_WEBHOOKS=true`; each integer `update_id` is claimed as `telegram:webhook-update:{sha256(bot_user_id)}:{update_id}` (24h) before dispatch; state/identity failure → 503. Python-surface adaptations only: the `ValidationError` text names options in snake_case (`secret_token`, `allow_unverified_webhooks=True`); the shared `getMe` is an `asyncio.Task` awaited through `asyncio.shield` so a cancelled webhook does not cancel it for other waiters; `update_id` claiming mirrors `Number.isInteger`: an integral JSON float (`7.0`, `7e0` — floats under `json.loads`, the number 7 in JS) is normalised to `int` and shares the `...:7` key, while `bool` (an `int` subclass) and fractional floats are not claimed. **Deliberate Python divergence:** a non-`bool` `allow_unverified_webhooks` (e.g. the string `"false"`) raises `ValidationError` instead of being coerced truthy — upstream's TS `boolean` type rules this out at compile time, Python has no such guard and `bool("false")` would silently fail open. `secret_token` now resolves with `??` semantics (`is not None`), replacing the earlier `or` fallback, so an explicit `""` no longer silently picks up the env secret. **Deliberate Python divergence (fixes an upstream bug):** the `getMe` username is cached and reapplied on a repeat `initialize()` — upstream's `ensureBotIdentity` returns early once `webhookScope` is set, so after `initialize` re-applies `chat.getUserName()` the Telegram username is never restored and `@real_bot` / `/cmd@real_bot` stop routing after a shutdown + re-initialize. `allow_unverified_webhooks` is the *last* `TelegramAdapterConfig` field (not alphabetical) so positional callers are not shifted. | Same | Not ported here: the Vercel Connect async `bot_token` resolver (#189) and the polling checkpoint keyed on the same scope (#227). Cross-instance dedupe on Postgres relies on #240, which makes `set_if_not_exists` reclaim expired rows. | -| Telegram inbound parsing, allowlist and typing (4.41 wave, #225) | **Parity** with upstream `4ee187ac` (#612), `2531a422` (#621, Telegram half), `0701679e` (#706, regex cache only), `54eea715` (#742), `53bf73db` (#752), `a0ba9868` (#835), `a18e7922` (#836) and the Telegram half of `b6fa24c6` (#865). Python-surface adaptations only, no behavior divergence beyond the mention-regex fold and the `allowed_user_ids` type check noted here: location coordinates and dice values go through `_js_number_str`, which renders a JSON number as JS `String()` does (`51.0` -> `"51"`, `1e-05` -> `"0.00001"`, `1.5e-07` -> `"1.5e-7"`), since `json.loads` keeps floats that JS prints differently; invoice amounts use `f"{amount / 10**e:.{e}f}"` for `toFixed(e)`; the mention pattern is `@{re.escape(name)}(?![A-Za-z0-9_-])` with `re.IGNORECASE`, because JS `\w` without the `u` flag is ASCII-only while Python's is Unicode (JS `/i` still folds non-ASCII letters, so `re.ASCII` is not used: `@ботик` matches `@БОТИК` on both sides; one residual difference is that Python's Unicode folding also equates the Kelvin sign with `k` and long s `ſ` with `s`, which JS's non-`u` canonicalization never maps onto ASCII, so `@` + U+212A + `bot` mentions a `kbot` in Python only); allowlist ids are stringified with the same `_js_number_str` (`456.0` -> `"456"`); a non-list `allowed_user_ids` (e.g. a bare `"123,456"` string, which Python would iterate per character, where upstream's `.map` throws) raises `ValidationError`; and the acting-user chain (`callback_query.from` -> `message_reaction.user` -> first non-`None` of `message`/`edited_message`/`channel_post`/`edited_channel_post` `.from`) uses `is not None` at every step; the typing action runs as an `asyncio` task created before `Chat.process_message`/`process_slash_command` schedules the handler task (JS fires the request synchronously; in Python both are tasks and run FIFO, so typing still goes first), is passed to `wait_until` when given and otherwise held in `_typing_tasks` so it is not garbage-collected (`disconnect()` cancels and awaits any still pending before closing the aiohttp session, so none reopens it); the download uses the shared aiohttp session with `ClientTimeout(total=30)` (covers the body read) and a running byte count over `response.content.iter_chunked`, raising `NetworkError`. It deliberately does **not** use `chat_sdk.shared.download` (#204): its HTTPS/public-address checks would break self-hosted Bot API servers, which upstream also exempts, and `read_attachment_body` decodes `Content-Encoding` itself while the shared aiohttp session already auto-decompresses. | Same | Not ported here: the media-group (album) path, which must also start typing and honour the allowlist, plus the `pollingGroup` allowlist check (#227); `reply_to_message` parsing and `mentionOnReply` (#228); `business_message.from` in the allowlist chain (#189). | +| Telegram webhook verification + `update_id` dedupe (4.41 wave, #224) | **Parity** with upstream `c4a359e7` (vercel/chat#858, chat@4.39), `1d2b78d9` (#799, chat@4.38) and the bot-identity scope helper from `7a1150ce` (#813): webhook mode requires `secret_token` unless `allow_unverified_webhooks` / `TELEGRAM_ALLOW_UNVERIFIED_WEBHOOKS=true`; each integer `update_id` is claimed as `telegram:webhook-update:{sha256(bot_user_id)}:{update_id}` (24h) before dispatch; state/identity failure → 503. Python-surface adaptations only: the `ValidationError` text names options in snake_case (`secret_token`, `allow_unverified_webhooks=True`); the shared `getMe` is an `asyncio.Task` awaited through `asyncio.shield` so a cancelled webhook does not cancel it for other waiters; `update_id` claiming mirrors `Number.isInteger`: an integral JSON float (`7.0`, `7e0` — floats under `json.loads`, the number 7 in JS) is normalised to `int` and shares the `...:7` key, while `bool` (an `int` subclass) and fractional floats are not claimed. **Deliberate Python divergence:** a non-`bool` `allow_unverified_webhooks` (e.g. the string `"false"`) raises `ValidationError` instead of being coerced truthy — upstream's TS `boolean` type rules this out at compile time, Python has no such guard and `bool("false")` would silently fail open. `secret_token` now resolves with `??` semantics (`is not None`), replacing the earlier `or` fallback, so an explicit `""` no longer silently picks up the env secret. **Deliberate Python divergence (fixes an upstream bug):** the `getMe` username is cached and reapplied on a repeat `initialize()` — upstream's `ensureBotIdentity` returns early once `webhookScope` is set, so after `initialize` re-applies `chat.getUserName()` the Telegram username is never restored and `@real_bot` / `/cmd@real_bot` stop routing after a shutdown + re-initialize. `allow_unverified_webhooks` is the *last* `TelegramAdapterConfig` field (not alphabetical) so positional callers are not shifted. | Same | Not ported here: the Vercel Connect async `bot_token` resolver (#189) and the polling checkpoint keyed on the same scope (#227, now ported: see the Telegram polling section above). Cross-instance dedupe on Postgres relies on #240, which makes `set_if_not_exists` reclaim expired rows. | +| Telegram inbound parsing, allowlist and typing (4.41 wave, #225) | **Parity** with upstream `4ee187ac` (#612), `2531a422` (#621, Telegram half), `0701679e` (#706, regex cache only), `54eea715` (#742), `53bf73db` (#752), `a0ba9868` (#835), `a18e7922` (#836) and the Telegram half of `b6fa24c6` (#865). Python-surface adaptations only, no behavior divergence beyond the mention-regex fold and the `allowed_user_ids` type check noted here: location coordinates and dice values go through `_js_number_str`, which renders a JSON number as JS `String()` does (`51.0` -> `"51"`, `1e-05` -> `"0.00001"`, `1.5e-07` -> `"1.5e-7"`), since `json.loads` keeps floats that JS prints differently; invoice amounts use `f"{amount / 10**e:.{e}f}"` for `toFixed(e)`; the mention pattern is `@{re.escape(name)}(?![A-Za-z0-9_-])` with `re.IGNORECASE`, because JS `\w` without the `u` flag is ASCII-only while Python's is Unicode (JS `/i` still folds non-ASCII letters, so `re.ASCII` is not used: `@ботик` matches `@БОТИК` on both sides; one residual difference is that Python's Unicode folding also equates the Kelvin sign with `k` and long s `ſ` with `s`, which JS's non-`u` canonicalization never maps onto ASCII, so `@` + U+212A + `bot` mentions a `kbot` in Python only); allowlist ids are stringified with the same `_js_number_str` (`456.0` -> `"456"`); a non-list `allowed_user_ids` (e.g. a bare `"123,456"` string, which Python would iterate per character, where upstream's `.map` throws) raises `ValidationError`; and the acting-user chain (`callback_query.from` -> `message_reaction.user` -> first non-`None` of `message`/`edited_message`/`channel_post`/`edited_channel_post` `.from`) uses `is not None` at every step; the typing action runs as an `asyncio` task created before `Chat.process_message`/`process_slash_command` schedules the handler task (JS fires the request synchronously; in Python both are tasks and run FIFO, so typing still goes first), is passed to `wait_until` when given and otherwise held in `_typing_tasks` so it is not garbage-collected (`disconnect()` cancels and awaits any still pending before closing the aiohttp session, so none reopens it); the download uses the shared aiohttp session with `ClientTimeout(total=30)` (covers the body read) and a running byte count over `response.content.iter_chunked`, raising `NetworkError`. It deliberately does **not** use `chat_sdk.shared.download` (#204): its HTTPS/public-address checks would break self-hosted Bot API servers, which upstream also exempts, and `read_attachment_body` decodes `Content-Encoding` itself while the shared aiohttp session already auto-decompresses. | Same | The media-group (album) path, its typing and the `pollingGroup` allowlist check landed in #227 (see the Telegram polling section above). Not ported here: `reply_to_message` parsing and `mentionOnReply` (#228); `business_message.from` in the allowlist chain (#189). | | WhatsApp `get_user` | Raises `ChatNotImplementedError` (`Chat.get_user` translates to "does not support get_user") | Not implemented upstream either (no `getUser` on the WhatsApp adapter) | WhatsApp Cloud API has no user lookup endpoint — phone numbers are the only stable identifier and there's no equivalent of `users.info` exposed to business apps. Documented explicitly so callers don't expect parity with Slack/Teams/Discord. | | Messenger `get_user` | Raising stub (`ChatNotImplementedError`); a Graph-API-backed impl is tracked as issue #132 | No `getUser` method on the Messenger adapter | **Parity — upstream Messenger has no user-lookup method**; the Python raising stub matches. (Meta's Graph API *could* back a real implementation, unlike WhatsApp — hence #132 stays open as an enhancement.) | | Linear agent sessions | **Complete** (5-PR wave, **#151** — Wave D done). All five landed on `main`: L1 agent-session types (`LinearAgentSessionThreadId`, `LinearAgentSessionCommentRawMessage`, `mode`/`kind`), L2 the `:s:{session}` thread-id encode/decode, L3 the webhook PARSE + routing (`_parse_message_from_agent_session_event`, `_handle_agent_session_event`), L4 the agent-activity EMIT path (`post_message`/`start_typing`/`stream` session branches as raw GraphQL — see the "Linear agent-activity emit" divergence row above), and **L5 (this change)**: the agent-session FETCH path (`fetch_messages` → `_fetch_agent_session_messages`, the `edit_message`/`delete_message` append-only guards, and `fetch_thread` `agentSessionId` metadata as raw GraphQL — see the "Linear agent-session fetch" divergence row above). | Full agent-sessions support (`adapter-linear` 4.27.0, `bc94f0a`): parses agent-session webhook events into messages, emits agent activity, fetches the session thread, and routes the agent-session thread id | Largest single gap from the 0.4.30 audit; pre-existing (present since 0.4.29). Closed across the 4.31 wave — tracked in **#151**. | diff --git a/src/chat_sdk/adapters/telegram/adapter.py b/src/chat_sdk/adapters/telegram/adapter.py index 81fc35f2..3e4fb166 100644 --- a/src/chat_sdk/adapters/telegram/adapter.py +++ b/src/chat_sdk/adapters/telegram/adapter.py @@ -140,6 +140,17 @@ TELEGRAM_DEFAULT_POLLING_LIMIT = 100 TELEGRAM_DEFAULT_POLLING_RETRY_DELAY_MS = 1000 TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS = 250 +# Incoming albums (vercel/chat#760): parts are buffered in state until no new +# part has arrived for the settle window, then emitted as one message. +TELEGRAM_INCOMING_MEDIA_GROUP_BUFFER_TTL_MS = 30_000 +TELEGRAM_INCOMING_MEDIA_GROUP_LOCK_TTL_MS = 5_000 +TELEGRAM_INCOMING_MEDIA_GROUP_RETRY_MS = 50 +TELEGRAM_INCOMING_MEDIA_GROUP_SETTLE_MS = 1_000 +# Albums hold at most 10 items; the incoming buffer keeps the newest 10. +TELEGRAM_MEDIA_GROUP_MAX = 10 +# Upper bound for both the request-failure backoff and the per-update retry +# backoff of the polling loop (vercel/chat#942). +TELEGRAM_POLLING_MAX_BACKOFF_MS = 30_000 # Telegram rejects unparseable MarkdownV2 with a 400 whose description reads # "Bad Request: can't parse entities: ..." ("caption entities" for media # captions). Matched case-insensitively as a substring, like upstream's @@ -885,6 +896,14 @@ def __init__(self, config: TelegramAdapterConfig | None = None) -> None: self._runtime_mode: TelegramRuntimeMode = "webhook" self._polling_task: asyncio.Task[None] | None = None self._polling_active: bool = False + # True only while the polling loop is parked in ``getUpdates`` or one + # of its sleeps: the awaits upstream's per-iteration AbortController + # reaches. ``stop_polling`` cancels the loop only then, so in-flight + # handlers and checkpoint writes finish first, as upstream. + self._polling_interruptible: bool = False + # Strong references to webhook album tasks (vercel/chat#760) so they + # are not garbage-collected mid-settle when no ``wait_until`` is given. + self._media_group_tasks: set[asyncio.Task[None]] = set() # Draft-id counter for native DM draft streaming (vercel/chat#340). # Seeded from wall-clock millis (mod int32 max) so concurrent bot @@ -1191,6 +1210,15 @@ async def _get_http_session(self) -> Any: async def disconnect(self) -> None: """Disconnect the adapter, stop polling, and close the shared HTTP session.""" + # Python-only: ``Chat.shutdown`` cancels in-flight handler tasks before + # disconnecting adapters, but an album still settling would dispatch + # a new handler afterwards (and keep ``stop_polling`` waiting on it). + # Cancel albums first; a polled album then stays in the checkpoint. + media_group_tasks = [task for task in self._media_group_tasks if not task.done()] + for task in media_group_tasks: + task.cancel() + if media_group_tasks: + await asyncio.wait(media_group_tasks) await self.stop_polling() # Python-only: a pending receipt-typing task would otherwise reopen the # shared aiohttp session via ``_get_http_session`` after it is closed. @@ -1256,16 +1284,27 @@ async def _run_polling() -> None: async def stop_polling(self) -> None: """Stop long-polling. - Cancels the polling task so that a blocked long-poll HTTP request - does not cause a ~30 s hang on shutdown. + Like upstream's ``AbortController``, this interrupts only a pending + ``getUpdates`` request or a polling sleep, so a blocked long poll does + not hang shutdown for ~30 s. When the loop is waiting for update + handlers or writing its checkpoint, that step finishes first and the + loop then exits, so the checkpoint stays consistent. """ if not self._polling_active: return self._polling_active = False - if self._polling_task and not self._polling_task.done(): - self._polling_task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await self._polling_task + task = self._polling_task + if task is not None and not task.done(): + if self._polling_interruptible: + task.cancel() + # ``asyncio.wait`` rather than ``await task``: the loop's own + # cancellation is expected, but cancelling *this* caller must + # still propagate (``suppress(CancelledError)`` would swallow it). + # Upstream parity: like ``stopPolling``, this waits for in-flight + # handlers, so a polled handler must not await it (deadlock). + await asyncio.wait({task}) + if not task.cancelled(): + task.result() self._polling_task = None self._logger.info("Telegram polling stopped") @@ -1280,45 +1319,323 @@ async def reset_webhook(self, drop_pending_updates: bool = False) -> None: {"dropPendingUpdates": drop_pending_updates}, ) + def _now_ms(self) -> int: + """Epoch milliseconds for album settle windows and polling retries. + + Epoch time (upstream ``Date.now()``), not monotonic, because + ``receivedAt`` / ``retryAt`` are persisted in state and compared after + a restart or on another instance. Overridable so tests can inject a + fake clock (together with :meth:`_sleep`). + """ + return int(time.time() * 1000) + + async def _sleep(self, delay_ms: float) -> None: + """Port of upstream ``sleep``; overridable so tests can fake time.""" + if delay_ms <= 0: + return + await asyncio.sleep(delay_ms / 1000) + + async def _polling_abortable(self, awaitable: Awaitable[_T]) -> _T: + """Await *awaitable* as a point where ``stop_polling`` may cancel the loop. + + The Python stand-in for upstream's per-iteration ``AbortController``, + whose signal reaches only ``getUpdates`` and the loop's sleeps. + """ + self._polling_interruptible = True + try: + return await awaitable + finally: + self._polling_interruptible = False + + def _polling_group(self, update: TelegramUpdate) -> str | None: + """Key shared by the polled parts of one album, or ``None``. + + Port of upstream ``pollingGroup``: parts are grouped by chat, forum + topic and ``media_group_id``. With an allowlist, a part from a user + outside it is not grouped (it is then dropped by :meth:`process_update` + without ever being buffered in the checkpoint). + """ + message = self._message_update(update) + if self._allowed_user_ids is not None: + sender = cast( + "TelegramUser | None", + None if message is None else (message.get("from_user") or message.get("from")), # type: ignore[call-overload] + ) + if sender is None or _js_number_str(sender.get("id")) not in self._allowed_user_ids: + return None + if message is None or not message.get("media_group_id"): + return None + return json.dumps([message["chat"]["id"], message.get("message_thread_id"), message.get("media_group_id")]) + + def _polling_group_key(self, update: TelegramUpdate) -> str: + group = self._polling_group(update) + return group if group is not None else str(update.get("update_id")) + + @staticmethod + async def _await_update_tasks(tasks: list[asyncio.Task[Any]]) -> None: + """Wait for every handler task of one update, then raise the first failure. + + ``Promise.allSettled`` + rethrow. ``asyncio.wait`` never cancels the + handler tasks, even if the polling loop itself is cancelled. + """ + if not tasks: + return + await asyncio.wait(tasks) + for task in tasks: + if task.cancelled(): + # Python-only: ``Chat.shutdown`` cancels in-flight handler + # tasks. Count that as a failure so the update stays in the + # checkpoint instead of being acknowledged unhandled. + raise RuntimeError("Telegram update handler was cancelled") + error = task.exception() + if error is not None: + raise error + + async def _process_polling_updates( + self, + updates: list[TelegramUpdate], + completed: set[int], + ) -> list[tuple[TelegramUpdate, Exception | None]]: + """Dispatch polled updates and wait for their handlers. + + Port of upstream ``processPollingUpdates``. The parts of one album + succeed or fail together; updates already in *completed* are not + dispatched again. Returns ``(update, error)`` pairs in ``update_id`` + order, where ``error`` is ``None`` for a handled update. + """ + groups: dict[str, list[TelegramUpdate]] = {} + for update in updates: + groups.setdefault(self._polling_group_key(update), []).append(update) + + async def run_update(update: TelegramUpdate) -> None: + if update["update_id"] in completed: + return + # The checkpoint, not core dedupe, deduplicates polled updates: a + # retried update must not be dropped as a duplicate (#942). + # Upstream parity: with the "queue" / "debounce" strategies a + # message's task settles once it is enqueued, and a queued + # handler's failure surfaces on the task that drains the queue, so + # the retry is attributed to that update (same in ``chat.ts``). + # Also upstream parity: a handler dispatched here is admitted work + # that ``stop_polling`` waits for, even when it starts after + # ``Chat.shutdown``'s (Python-only) cancellation sweep. + await self._await_update_tasks(self.process_update(update, WebhookOptions(deduplicate=False))) + + async def run_group(group: list[TelegramUpdate]) -> list[tuple[TelegramUpdate, Exception | None]]: + try: + results = await asyncio.gather(*(run_update(update) for update in group), return_exceptions=True) + for result in results: + if isinstance(result, BaseException): + raise result + for update in group: + completed.add(update["update_id"]) + return [(update, None) for update in group] + except Exception as error: + return [(update, error) for update in group] + + grouped = await asyncio.gather(*(run_group(group) for group in groups.values())) + return sorted( + (result for results in grouped for result in results), + key=lambda result: result[0]["update_id"], + ) + + async def _fetch_polling_updates(self, params: dict[str, Any], collect_ms: int | None) -> list[TelegramUpdate]: + """Call ``getUpdates``, giving up with no updates after *collect_ms*. + + Upstream aborts the long poll with a timer (``collecting``) once a + pending album has settled or a retry falls due. Abandoned updates are + not acknowledged, so Telegram returns them again on the next poll. + """ + if collect_ms is None or collect_ms <= 0: + return await self.telegram_fetch("getUpdates", params) + fetch = asyncio.ensure_future(self.telegram_fetch("getUpdates", params)) + timer = asyncio.ensure_future(self._sleep(collect_ms)) + try: + await asyncio.wait({fetch, timer}, return_when=asyncio.FIRST_COMPLETED) + finally: + timer.cancel() + if not fetch.done(): + fetch.cancel() + # A cancelled fetch only finishes cancelling once it runs again. + await asyncio.wait({fetch}) + if fetch.cancelled(): + return [] + return fetch.result() + async def polling_loop(self, config: ResolvedTelegramLongPollingConfig) -> None: - """Core polling loop that calls ``getUpdates`` in a loop.""" - offset: int | None = None + """Long-poll ``getUpdates``, acknowledging only settled updates. + + Port of upstream ``pollingLoop`` (vercel/chat#942). Each batch's + handler tasks are awaited before ``offset`` moves past it. A failed + update (or every part of a failed album) is saved in the + ``telegram:polling:{scope}`` checkpoint with exponential backoff and + retried; album parts wait there for the settle window and are then + dispatched together. ``stop_polling`` cancels the loop only while it + waits in ``getUpdates`` or a sleep (see :meth:`_polling_abortable`). + """ + if not self._chat: + return + state = self._chat.get_state() + checkpoint: dict[str, Any] | None = None consecutive_failures = 0 - max_backoff_ms = 30_000 + drained = False + completed: set[int] = set() + + def retry(entries: list[dict[str, Any]], error: Exception) -> dict[str, Any]: + attempts = max((entry.get("attempts") if entry.get("attempts") is not None else 0) for entry in entries) + 1 + retry_after = error.retry_after if isinstance(error, AdapterRateLimitError) else None + delay = max( + min( + max(config.retry_delay_ms, TELEGRAM_DEFAULT_POLLING_RETRY_DELAY_MS) * 2 ** min(attempts - 1, 30), + TELEGRAM_POLLING_MAX_BACKOFF_MS, + ), + retry_after * 1000 if retry_after is not None else 0, + ) + retry_at = self._now_ms() + int(delay) + self._logger.warn( + "Telegram polling update processing failed", + { + "error": str(error), + "updateId": entries[0]["update"].get("update_id"), + "attempts": attempts, + "retryAt": retry_at, + }, + ) + return {"attempts": attempts, "retryAt": retry_at} + + def deadline(entries: list[dict[str, Any]]) -> int: + return max( + max( + entry["receivedAt"] + + (TELEGRAM_INCOMING_MEDIA_GROUP_SETTLE_MS if self._polling_group(entry["update"]) else 0), + entry["retryAt"] if entry.get("retryAt") is not None else 0, + ) + for entry in entries + ) while self._polling_active: try: - params: dict[str, Any] = {"limit": config.limit, "timeout": config.timeout} + # Retried every iteration, so a failed startup ``getMe`` + # recovers once Telegram is reachable. + await self._ensure_bot_identity() + key = f"{self._name}:polling:{self._webhook_scope}" + if checkpoint is None: + stored: dict[str, Any] | None = await state.get(key) + checkpoint = stored if stored is not None else {"pending": []} + offset: int | None = checkpoint.get("offset") + acknowledged = offset if offset is not None else 0 + + groups: dict[str, list[dict[str, Any]]] = {} + for entry in checkpoint["pending"]: + groups.setdefault(self._polling_group_key(entry["update"]), []).append(entry) + eligible = [ + entries + for entries in groups.values() + if all(entry["update"]["update_id"] < acknowledged for entry in entries) + ] + ready = [entries for entries in eligible if deadline(entries) <= self._now_ms()] + if not self._polling_active: + # Python-specific divergence: stopped during the identity + # or checkpoint await, so dispatch nothing new. Upstream + # has no check here and still runs the ready retry batch; + # here it could start handlers after ``Chat.shutdown``'s + # cancellation sweep (``disconnect`` has already cancelled + # settling albums). The batch stays in the checkpoint. + return + if ready and not drained: + results = await self._process_polling_updates( + [entry["update"] for entries in ready for entry in entries], + completed, + ) + failures = {update["update_id"]: error for update, error in results if error is not None} + retries: dict[int, dict[str, Any]] = {} + for entries in ready: + failed = next((entry for entry in entries if entry["update"]["update_id"] in failures), None) + if failed is None: + continue + scheduled = retry(entries, failures[failed["update"]["update_id"]]) + for entry in entries: + retries[entry["update"]["update_id"]] = scheduled + pending = [ + {**entry, **retries.get(entry["update"]["update_id"], {})} + for entry in checkpoint["pending"] + if entry["update"]["update_id"] not in completed + ] + next_checkpoint: dict[str, Any] = {**checkpoint, "pending": pending} + if pending: + await state.set(key, next_checkpoint) + else: + await state.delete(key) + checkpoint = next_checkpoint + for update, _error in results: + completed.discard(update["update_id"]) + consecutive_failures = 0 + # Poll once before the next ready batch, so new updates + # are not starved by a run of retries. + drained = True + continue + + remaining = ( + max(0, min(deadline(entries) for entries in eligible) - self._now_ms()) if eligible else None + ) + if not self._polling_active: + # Stopped while the loop could not be cancelled; upstream's + # already-aborted signal fails this fetch at once. + return + params: dict[str, Any] = { + "limit": config.limit, + # Ask for an immediate answer when a batch is already due. + "timeout": 0 if remaining == 0 else config.timeout, + } if offset is not None: params["offset"] = offset if config.allowed_updates is not None: params["allowed_updates"] = config.allowed_updates - updates: list[TelegramUpdate] = await self.telegram_fetch( - "getUpdates", - params, + updates: list[TelegramUpdate] = await self._polling_abortable( + self._fetch_polling_updates(params, remaining) ) + drained = False + if not self._polling_active: + return - consecutive_failures = 0 - + results = await self._process_polling_updates( + [update for update in updates if self._polling_group(update) is None], + completed, + ) + pending_by_id: dict[int, dict[str, Any]] = { + entry["update"]["update_id"]: entry for entry in checkpoint["pending"] + } + for update, error in results: + if error is not None: + failed_entry: dict[str, Any] = {"update": update, "receivedAt": self._now_ms()} + pending_by_id[update["update_id"]] = {**failed_entry, **retry([failed_entry], error)} for update in updates: - offset = update.get("update_id", 0) + 1 - try: - self.process_update(update) - except Exception as error: - self._logger.warn( - "Failed to process Telegram polled update", - { - "error": str(error), - "updateId": update.get("update_id"), - }, - ) - except asyncio.CancelledError: - return + update_id = update["update_id"] + if ( + self._polling_group(update) is not None + and update_id not in completed + and update_id not in pending_by_id + ): + pending_by_id[update_id] = {"update": update, "receivedAt": self._now_ms()} + next_offset = offset + for update in updates: + next_offset = update["update_id"] + 1 + next_checkpoint: dict[str, Any] = {"offset": next_offset, "pending": list(pending_by_id.values())} + # Saved before the next ``getUpdates`` acknowledges these + # updates; a failed write keeps the old offset in memory too. + if next_checkpoint["pending"] or checkpoint["pending"]: + await state.set(key, next_checkpoint) + checkpoint = next_checkpoint + acknowledged = next_offset if next_offset is not None else 0 + completed.difference_update([update_id for update_id in completed if update_id < acknowledged]) + if not updates and remaining is not None and self._polling_active: + await self._polling_abortable(self._sleep(min(remaining, TELEGRAM_INCOMING_MEDIA_GROUP_RETRY_MS))) + consecutive_failures = 0 except Exception as error: consecutive_failures += 1 backoff_ms = min( config.retry_delay_ms * 2 ** (consecutive_failures - 1), - max_backoff_ms, + TELEGRAM_POLLING_MAX_BACKOFF_MS, ) self._logger.warn( @@ -1333,7 +1650,7 @@ async def polling_loop(self, config: ResolvedTelegramLongPollingConfig) -> None: if not self._polling_active: return - await asyncio.sleep(backoff_ms / 1000.0) + await self._polling_abortable(self._sleep(backoff_ms)) # -- Runtime mode resolution --------------------------------------------- @@ -1393,15 +1710,13 @@ async def _fetch_webhook_info(self) -> TelegramWebhookInfo | None: # -- Update dispatching -------------------------------------------------- - def process_update( - self, - update: TelegramUpdate, - options: WebhookOptions | None = None, - ) -> None: - """Dispatch a Telegram update to the appropriate handler.""" - # ``message ?? edited_message ?? channel_post ?? edited_channel_post``: - # the first present (non-None) payload, as upstream's nullish chain. - message_update: TelegramMessage | None = next( + @staticmethod + def _message_update(update: TelegramUpdate) -> TelegramMessage | None: + """``message ?? edited_message ?? channel_post ?? edited_channel_post``. + + The first present (non-None) payload, as upstream's nullish chain. + """ + return next( ( candidate for candidate in ( @@ -1415,30 +1730,59 @@ def process_update( None, ) + def process_update( + self, + update: TelegramUpdate, + options: WebhookOptions | None = None, + ) -> list[asyncio.Task[Any]]: + """Dispatch a Telegram update to the appropriate handler. + + Returns every dispatched handler task (message, album, slash command, + action, reactions) so the polling loop can wait for them before + acknowledging the update (vercel/chat#942). The webhook path ignores + the list. + """ + tasks: list[asyncio.Task[Any]] = [] + message_update = self._message_update(update) + # User allowlist (vercel/chat#742), checked before any routing. With an # allowlist set, an update with no acting user (e.g. an anonymous # channel post) is dropped too. if self._allowed_user_ids is not None: user_id = self._update_user_id(update, message_update) if user_id is None or _js_number_str(user_id) not in self._allowed_user_ids: - return + return tasks # Slash commands are gated to fresh ``message`` updates only — edited # messages and channel posts never route to the slash-command - # handlers. ``handle_slash_command_update`` returns ``True`` when it - # consumed the update, in which case the regular message path is - # skipped (mirrors upstream ``messageUpdate && !handledSlashCommand``). + # handlers, and neither do album parts (vercel/chat#760), whose + # caption is the album's text. ``handle_slash_command_update`` returns + # a truthy value when it consumed the update, in which case the + # regular message path is skipped (upstream + # ``messageUpdate && !handledSlashCommand``). message = update.get("message") - handled_slash_command = message is not None and self.handle_slash_command_update(message, options) + handled_slash_command: asyncio.Task[Any] | bool = ( + message is not None + and not message.get("media_group_id") + and self.handle_slash_command_update(message, options) + ) + if not isinstance(handled_slash_command, bool): + tasks.append(handled_slash_command) if message_update is not None and not handled_slash_command: - self.handle_incoming_message_update(message_update, options) + message_task = self.handle_incoming_message_update(message_update, options) + if message_task is not None: + tasks.append(message_task) if update.get("callback_query"): - self.handle_callback_query(update["callback_query"], options) + action_task = self.handle_callback_query(update["callback_query"], options) + if action_task is not None: + tasks.append(action_task) if update.get("message_reaction"): - self.handle_message_reaction_update(update["message_reaction"], options) + tasks.extend(self.handle_message_reaction_update(update["message_reaction"], options)) + + return tasks @staticmethod def _update_user_id(update: TelegramUpdate, message_update: TelegramMessage | None) -> Any: @@ -1466,10 +1810,14 @@ def handle_incoming_message_update( self, telegram_message: TelegramMessage, options: WebhookOptions | None = None, - ) -> None: - """Handle a new or edited message update.""" + ) -> asyncio.Task[Any] | None: + """Handle a new or edited message update. + + Returns the handler task, or for an album part the task that buffers + it and, once the album has settled, dispatches the combined message. + """ if not self._chat: - return + return None thread_id = self.encode_thread_id( TelegramThreadId( @@ -1478,23 +1826,181 @@ def handle_incoming_message_update( ) ) + if telegram_message.get("media_group_id"): + return self._start_incoming_media_group(telegram_message, thread_id, options) + self._start_typing_for_private_message(telegram_message, thread_id, options) parsed_message = self.parse_telegram_message(telegram_message, thread_id) self.cache_message(parsed_message) - self._chat.process_message(self, thread_id, parsed_message, options) + return self._chat.process_message(self, thread_id, parsed_message, options) + + def _start_incoming_media_group( + self, + telegram_message: TelegramMessage, + thread_id: str, + options: WebhookOptions | None, + ) -> asyncio.Task[None] | None: + """Run :meth:`_process_incoming_media_group` as a task. + + A failure is logged; ``wait_until`` gets a wrapper that settles + without raising (upstream's ``task.catch(log)``), while the returned + task still raises so the polling loop can retry the album. + """ + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return None + media_group_id = telegram_message.get("media_group_id") + task = loop.create_task(self._process_incoming_media_group(telegram_message, thread_id, options)) + self._media_group_tasks.add(task) + + def _done(done: asyncio.Task[None]) -> None: + self._media_group_tasks.discard(done) + if done.cancelled(): + return + error = done.exception() + if error is not None: + self._logger.warn( + "Failed to process incoming Telegram media group", + {"error": str(error), "mediaGroupId": media_group_id, "threadId": thread_id}, + ) + + task.add_done_callback(_done) + + if options is not None and options.wait_until is not None: + + async def _settled() -> None: + # Errors are logged by ``_done``; cancelling this wrapper does + # not cancel the album task. Like ``Chat._tracked``, an album + # cancelled by ``disconnect()`` counts as settled; only a + # cancellation of this wrapper itself propagates. + try: + await asyncio.shield(task) + except asyncio.CancelledError: + current = asyncio.current_task() + if current is not None and current.cancelling(): + raise + except Exception: # noqa: S110 — logged by ``_done`` + pass + + options.wait_until(loop.create_task(_settled())) + return task + + async def _process_incoming_media_group( + self, + telegram_message: TelegramMessage, + thread_id: str, + options: WebhookOptions | None = None, + ) -> None: + """Buffer one album part and dispatch the album once it has settled. + + Port of upstream ``processIncomingMediaGroup`` (vercel/chat#760). + Parts are appended under a state lock to + ``telegram:incoming-media-group:{thread}:{media_group_id}`` (shared + across instances). Each part's task waits until no part has arrived + for the settle window; the first to see it settled removes the buffer + and dispatches one message carrying the newest part's id and raw, the + first non-empty text, every attachment in ``message_id`` order and + ``is_mention`` if any part mentions the bot. The others return. + """ + media_group_id = telegram_message.get("media_group_id") + if not (self._chat and media_group_id): + return + + state = self._chat.get_state() + # Upstream parity: the buffer key is not scoped to the bot identity + # (upstream ``processIncomingMediaGroup`` uses the same key), so bots + # sharing one state namespace must not share a group chat. + media_group_key = f"{self._name}:incoming-media-group:{thread_id}:{media_group_id}" + lock_key = f"{media_group_key}:lock" + appended = False + + while True: + lock = await state.acquire_lock(lock_key, TELEGRAM_INCOMING_MEDIA_GROUP_LOCK_TTL_MS) + if not lock: + await self._sleep(TELEGRAM_INCOMING_MEDIA_GROUP_RETRY_MS) + continue + + entries: list[dict[str, Any]] = [] + remaining_settle_ms = 0 + try: + stored = await state.get(media_group_key) + entries = list(stored) if stored is not None else [] + if not appended: + entries = [ + entry for entry in entries if entry["message"]["message_id"] != telegram_message["message_id"] + ] + entries.append({"message": telegram_message, "receivedAt": self._now_ms()}) + await state.set( + media_group_key, + entries[-TELEGRAM_MEDIA_GROUP_MAX:], + TELEGRAM_INCOMING_MEDIA_GROUP_BUFFER_TTL_MS, + ) + appended = True + remaining_settle_ms = TELEGRAM_INCOMING_MEDIA_GROUP_SETTLE_MS + elif not entries: + # Already dispatched by another part, or (upstream parity) + # the buffer expired after a stall longer than its TTL. + return + else: + newest_received_at = max(entry["receivedAt"] for entry in entries) + remaining_settle_ms = max( + 0, + TELEGRAM_INCOMING_MEDIA_GROUP_SETTLE_MS - (self._now_ms() - newest_received_at), + ) + if remaining_settle_ms == 0: + await state.delete(media_group_key) + finally: + await state.release_lock(lock) + + if remaining_settle_ms > 0: + await self._sleep(remaining_settle_ms) + continue + + ordered_messages = sorted( + (cast("TelegramMessage", entry["message"]) for entry in entries), + key=lambda message: message["message_id"], + ) + parsed_messages = [self.parse_telegram_message(message, thread_id) for message in ordered_messages] + if not parsed_messages: + return + latest_message = parsed_messages[-1] + content_message = next((message for message in parsed_messages if message.text), latest_message) + links = [link for message in parsed_messages for link in (message.links or [])] + combined_message = Message( + id=latest_message.id, + thread_id=thread_id, + text=content_message.text, + formatted=content_message.formatted, + raw=latest_message.raw, + author=latest_message.author, + metadata=latest_message.metadata, + attachments=[attachment for message in parsed_messages for attachment in message.attachments], + # Telegram parsing populates ``reply_to`` from #228. + reply_to=next((message.reply_to for message in parsed_messages if message.reply_to is not None), None), + is_mention=any(message.is_mention for message in parsed_messages), + links=links or None, + ) + + self._start_typing_for_private_message(ordered_messages[0], thread_id) + self.cache_message(combined_message) + task = self._chat.process_message(self, thread_id, combined_message, options) + if task is not None: + await task + return def handle_slash_command_update( self, telegram_message: TelegramMessage, options: WebhookOptions | None = None, - ) -> bool: + ) -> asyncio.Task[Any] | bool: """Route a leading ``/command`` message to the slash-command handlers. - Returns ``True`` when the update was consumed as a slash command (so - :meth:`process_update` skips the regular message path), and ``False`` - otherwise. Like the Discord adapter, the event is built with + Returns the handler task (``True`` without a running loop) when the + update was consumed as a slash command, so :meth:`process_update` + skips the regular message path, and ``False`` otherwise. Like the Discord adapter, the event is built with ``channel=None`` and the resolved thread ID is attached as ``channel_id`` — ``Chat`` re-wraps it into a real ``Channel`` before invoking handlers. @@ -1527,9 +2033,9 @@ def handle_slash_command_update( raw=telegram_message, ) event.channel_id = thread_id # type: ignore[attr-defined] - self._chat.process_slash_command(event, options) + task = self._chat.process_slash_command(event, options) - return True + return task if task is not None else True def _start_typing_for_private_message( self, @@ -1629,10 +2135,13 @@ def handle_callback_query( self, callback_query: TelegramCallbackQuery, options: WebhookOptions | None = None, - ) -> None: - """Handle a callback query (inline keyboard button press).""" + ) -> asyncio.Task[Any] | None: + """Handle a callback query (inline keyboard button press). + + Returns the action handler task; the acknowledgement is not part of it. + """ if not (self._chat and callback_query.get("message")): - return + return None message = callback_query["message"] thread_id = self.encode_thread_id( @@ -1668,7 +2177,7 @@ def handle_callback_query( ) ) - self._chat.process_action( + action_task = self._chat.process_action( ActionEvent( adapter=self, thread=None, # pyrefly: ignore[bad-argument-type] # filled in by Chat @@ -1702,15 +2211,17 @@ async def _ack() -> None: task = None if task and callable(wait_until): wait_until(task) + return action_task def handle_message_reaction_update( self, reaction_update: TelegramMessageReactionUpdated, options: WebhookOptions | None = None, - ) -> None: - """Handle a message reaction update.""" + ) -> list[asyncio.Task[Any]]: + """Handle a message reaction update; returns the reaction handler tasks.""" + tasks: list[asyncio.Task[Any]] = [] if not self._chat: - return + return tasks thread_id = self.encode_thread_id( TelegramThreadId( @@ -1733,7 +2244,7 @@ def handle_message_reaction_update( for reaction in reaction_update.get("new_reaction", []): key = self.reaction_key(reaction) if key not in old_reactions: - self._chat.process_reaction( + added_task = self._chat.process_reaction( ReactionEvent( adapter=self, thread=None, # pyrefly: ignore[bad-argument-type] # filled in by Chat @@ -1747,11 +2258,13 @@ def handle_message_reaction_update( ), options, ) + if added_task is not None: + tasks.append(added_task) for reaction in reaction_update.get("old_reaction", []): key = self.reaction_key(reaction) if key not in new_reactions: - self._chat.process_reaction( + removed_task = self._chat.process_reaction( ReactionEvent( adapter=self, thread=None, # pyrefly: ignore[bad-argument-type] # filled in by Chat @@ -1765,6 +2278,9 @@ def handle_message_reaction_update( ), options, ) + if removed_task is not None: + tasks.append(removed_task) + return tasks # -- Posting / editing / deleting ---------------------------------------- diff --git a/src/chat_sdk/adapters/telegram/types.py b/src/chat_sdk/adapters/telegram/types.py index d8215430..2815bff3 100644 --- a/src/chat_sdk/adapters/telegram/types.py +++ b/src/chat_sdk/adapters/telegram/types.py @@ -804,6 +804,8 @@ class TelegramMessage(TypedDict, total=False): game: TelegramGame invoice: TelegramInvoice location: TelegramLocation + # Album (media group) id shared by every part of one album. + media_group_id: str message_id: int # required message_thread_id: int photo: list[TelegramPhotoSize] diff --git a/tests/test_telegram_webhook.py b/tests/test_telegram_webhook.py index b823d5ac..46c353d7 100644 --- a/tests/test_telegram_webhook.py +++ b/tests/test_telegram_webhook.py @@ -10,6 +10,7 @@ import asyncio import hashlib +import inspect import json import os from dataclasses import dataclass @@ -26,10 +27,12 @@ create_telegram_adapter, ) from chat_sdk.adapters.telegram.cards import encode_telegram_callback_data -from chat_sdk.adapters.telegram.types import TelegramAdapterConfig, TelegramThreadId -from chat_sdk.shared.errors import NetworkError, ValidationError -from chat_sdk.shared.mock_adapter import MockStateAdapter, create_mock_state -from chat_sdk.types import Message, WebhookOptions +from chat_sdk.adapters.telegram.types import TelegramAdapterConfig, TelegramLongPollingConfig, TelegramThreadId +from chat_sdk.chat import Chat +from chat_sdk.shared.errors import AdapterRateLimitError, NetworkError, ValidationError +from chat_sdk.shared.mock_adapter import MockLogger, MockStateAdapter, create_mock_state +from chat_sdk.state.memory import MemoryStateAdapter +from chat_sdk.types import ChatConfig, LinkPreview, Message, WebhookOptions # --------------------------------------------------------------------------- # Helpers @@ -2361,3 +2364,1022 @@ async def __aexit__(self, *_exc: Any) -> None: with pytest.raises(NetworkError, match="Failed to download Telegram file f1") as info: await adapter.download_file("f1") assert isinstance(info.value.original_error, TimeoutError) + + +# --------------------------------------------------------------------------- +# Albums and polling acknowledgement (vercel/chat#760, #942) +# --------------------------------------------------------------------------- + +_POLLING_CHECKPOINT = f"telegram:polling:{hashlib.sha256(b'999').hexdigest()}" + + +async def _flush(rounds: int = 200) -> None: + """Let every runnable task advance, without wall-clock sleeping.""" + for _ in range(rounds): + await asyncio.sleep(0) + + +async def _wait_until(condition: Any, rounds: int = 5000) -> None: + """Yield to the event loop until ``condition()`` (sync or async) holds.""" + for _ in range(rounds): + result = condition() + if inspect.isawaitable(result): + result = await result + if result: + return + await asyncio.sleep(0) + raise AssertionError("condition was never met") + + +class _FakeClock: + """``vi.useFakeTimers`` stand-in for the adapter's ``_now_ms`` / ``_sleep``.""" + + def __init__(self, now: int = 1_700_000_000_000) -> None: + self.now = now + self._sleepers: list[tuple[int, asyncio.Future[None]]] = [] + + def now_ms(self) -> int: + return self.now + + async def sleep(self, delay_ms: float) -> None: + if delay_ms <= 0: + return + future: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self._sleepers.append((self.now + int(delay_ms), future)) + await future + + def install(self, adapter: TelegramAdapter) -> None: + adapter._now_ms = self.now_ms # type: ignore[method-assign] + adapter._sleep = self.sleep # type: ignore[method-assign] + + async def advance(self, delay_ms: int) -> None: + """Advance time, waking each sleeper at its deadline in order.""" + target = self.now + delay_ms + while True: + await _flush() + due = [wake for wake, future in self._sleepers if wake <= target and not future.done()] + if not due: + break + self.now = max(self.now, min(due)) + for wake, future in self._sleepers: + if wake <= self.now and not future.done(): + future.set_result(None) + self._sleepers = [(wake, future) for wake, future in self._sleepers if not future.done()] + self.now = target + await _flush() + + +class _FakeTelegramApi: + """Bot API stand-in whose ``getUpdates`` honours ``offset`` / ``limit``. + + Like Telegram, it keeps returning every update at or after ``offset`` + (unacknowledged updates come back), and with nothing to return it blocks + as a long poll until :meth:`deliver` or cancellation. + """ + + def __init__(self, updates: list[dict[str, Any]] | None = None, bot_id: int = 999) -> None: + self.updates = list(updates or []) + self.bot_id = bot_id + self.calls: list[tuple[str, Any]] = [] + self.polls: list[dict[str, Any]] = [] + self._arrived = asyncio.Event() + + async def __call__(self, method: str, payload: Any = None, **_kwargs: Any) -> Any: + self.calls.append((method, payload)) + if method == "getMe": + return {"id": self.bot_id, "is_bot": True, "first_name": "Bot", "username": "mybot"} + if method != "getUpdates": + return True + self.polls.append(dict(payload)) + offset = payload.get("offset", 0) + while True: + pending = [update for update in self.updates if update["update_id"] >= offset][: payload["limit"]] + if pending or payload["timeout"] == 0: + return pending + self._arrived.clear() + await self._arrived.wait() + + def deliver(self, update: dict[str, Any]) -> None: + self.updates.append(update) + self._arrived.set() + + +def _polled_message(update_id: int, *, chat_id: int = 1, **overrides: Any) -> dict[str, Any]: + message: dict[str, Any] = { + "message_id": update_id, + "date": 1, + "text": "hello", + "chat": {"id": chat_id, "type": "private"}, + "from": {"id": 2, "is_bot": False, "first_name": "User"}, + } + message.update(overrides) + return {"update_id": update_id, "message": message} + + +def _polled_album(*, chat_id: int = 1, first_id: int = 1) -> list[dict[str, Any]]: + parts = [ + _polled_message( + update_id, + chat_id=chat_id, + media_group_id="album", + photo=[{"file_id": str(update_id), "file_unique_id": str(update_id), "width": 10, "height": 10}], + ) + for update_id in (first_id, first_id + 1) + ] + for part in parts: + del part["message"]["text"] + return parts + + +def _polled_kind(kind: str) -> dict[str, Any]: + """Update ``1`` as an ordinary message, slash command, button action or reaction.""" + ordinary = _polled_message(1) + if kind == "message": + return ordinary + if kind == "command": + return _polled_message(1, text="/help", entities=[{"type": "bot_command", "offset": 0, "length": 5}]) + if kind == "action": + return { + "update_id": 1, + "callback_query": { + "id": "callback", + "chat_instance": "instance", + "from": ordinary["message"]["from"], + "message": ordinary["message"], + "data": "approve", + }, + } + return { + "update_id": 1, + "message_reaction": { + "chat": ordinary["message"]["chat"], + "message_id": 1, + "date": 1, + "user": ordinary["message"]["from"], + "old_reaction": [], + "new_reaction": [{"type": "custom_emoji", "custom_emoji_id": "reaction"}], + }, + } + + +class _PollingFixture: + """Real ``Chat`` + in-memory state + a Telegram adapter on a fake Bot API.""" + + def __init__( + self, + updates: list[dict[str, Any]] | None = None, + *, + state: MemoryStateAdapter | None = None, + clock: _FakeClock | None = None, + bot_id: int = 999, + allowed_user_ids: list[int | str] | None = None, + ) -> None: + self.api = _FakeTelegramApi(updates, bot_id) + self.state = state if state is not None else MemoryStateAdapter() + self.clock = clock if clock is not None else _FakeClock() + self.adapter = _make_adapter( + mode="webhook", + secret_token="secret", + user_name="mybot", + logger=MockLogger(), + allowed_user_ids=allowed_user_ids, + ) + self.adapter.telegram_fetch = self.api # type: ignore[method-assign] + self.clock.install(self.adapter) + self.chat = Chat( + ChatConfig( + user_name="mybot", + adapters={"telegram": self.adapter}, + state=self.state, + logger=MockLogger(), + concurrency="concurrent", + ) + ) + + async def start(self, limit: int = 100, retry_delay_ms: int = 10) -> None: + await self.chat.initialize() + await self.adapter.start_polling( + TelegramLongPollingConfig(limit=limit, retry_delay_ms=retry_delay_ms, timeout=1) + ) + + async def stop(self) -> None: + # ``stop_polling`` lets an in-flight batch finish, and an album in + # that batch settles on the fake clock, so keep time moving. + stopping = asyncio.ensure_future(self.adapter.stop_polling()) + for _ in range(100): + if stopping.done(): + break + await self.clock.advance(1_000) + await stopping + await self.chat.shutdown() + + async def checkpoint(self) -> Any: + await self.state.connect() + return await self.state.get(_POLLING_CHECKPOINT) + + +class TestTelegramIncomingMediaGroup: + """Webhook album parts are buffered and dispatched as one message (#760).""" + + @pytest.mark.asyncio + async def test_combines_an_incoming_media_group_into_one_ordered_message(self): + clock = _FakeClock() + state = create_mock_state() + adapters: list[TelegramAdapter] = [] + chats: list[MagicMock] = [] + for _ in range(2): + adapter = _make_adapter(mode="webhook", secret_token="secret", user_name="mybot") + adapter.telegram_fetch = AsyncMock( # type: ignore[method-assign] + return_value={"id": 999, "is_bot": True, "first_name": "Bot", "username": "mybot"} + ) + clock.install(adapter) + chat = MagicMock() + chat.get_state.return_value = state + chat.get_user_name.return_value = "mybot" + chat.process_message = MagicMock( + side_effect=lambda *_args, **_kwargs: asyncio.get_running_loop().create_task(asyncio.sleep(0)) + ) + await adapter.initialize(chat) + adapters.append(adapter) + chats.append(chat) + + pending: list[Any] = [] + messages = [ + _sample_message( + message_id=42, + media_group_id="meal-album", + text=None, + photo=[{"file_id": "photo-2", "file_unique_id": "photo-unique-2", "width": 800, "height": 600}], + ), + _sample_message( + message_id=41, + media_group_id="meal-album", + text=None, + caption="/analyze both pieces", + caption_entities=[{"type": "bot_command", "offset": 0, "length": 8}], + photo=[{"file_id": "photo-1", "file_unique_id": "photo-unique-1", "width": 800, "height": 600}], + ), + ] + for index, message in enumerate(messages): + message.pop("text") + response = await adapters[index].handle_webhook( + _make_request(json.dumps({"update_id": index + 1, "message": message}), secret_token="secret"), + WebhookOptions(wait_until=pending.append), + ) + assert response["status"] == 200 + + await _flush() + assert all(chat.process_message.call_count == 0 for chat in chats) + + await clock.advance(1_000) + await asyncio.gather(*pending) + + calls = [call for chat in chats for call in chat.process_message.call_args_list] + assert len(calls) == 1 + _adapter, thread_id, parsed, _options = calls[0].args + assert thread_id == "telegram:123" + assert parsed.id == "123:42" + assert parsed.text == "/analyze both pieces" + assert [attachment.fetch_metadata["fileId"] for attachment in parsed.attachments] == ["photo-1", "photo-2"] + # An album caption is the album's text, never a slash command. + assert all(chat.process_slash_command.call_count == 0 for chat in chats) + assert await state.get("telegram:incoming-media-group:telegram:123:meal-album") is None + + @pytest.mark.asyncio + async def test_album_handler_failure_is_logged_and_waituntil_settles(self): + """Python-specific: ``wait_until`` gets a settling wrapper, the returned task still raises.""" + clock = _FakeClock() + adapter = _make_adapter(mode="webhook", secret_token="secret", user_name="mybot", logger=MockLogger()) + clock.install(adapter) + chat = MagicMock() + chat.get_state.return_value = create_mock_state() + + async def _fail() -> None: + raise RuntimeError("handler failed") + + chat.process_message = MagicMock( + side_effect=lambda *_args, **_kwargs: asyncio.get_running_loop().create_task(_fail()) + ) + adapter._chat = chat + waited: list[Any] = [] + message = _sample_message(chat={"id": -100, "type": "group"}, media_group_id="g", caption="c") + message.pop("text") + + tasks = adapter.process_update({"update_id": 1, "message": message}, WebhookOptions(wait_until=waited.append)) # type: ignore[typeddict-item] + await clock.advance(1_000) + + assert len(tasks) == 1 + with pytest.raises(RuntimeError, match="handler failed"): + await tasks[0] + assert len(waited) == 1 + assert await waited[0] is None + assert ( + "Failed to process incoming Telegram media group", + {"error": "handler failed", "mediaGroupId": "g", "threadId": "telegram:-100"}, + ) in adapter._logger.warn.calls # type: ignore[attr-defined] + + @pytest.mark.asyncio + async def test_combined_album_keeps_the_newest_ten_parts_once_each_and_merges_their_fields(self): + """Newest-10 cap, redelivery dedupe, ``is_mention`` from any part, links and ``reply_to``.""" + clock = _FakeClock() + adapter = _make_adapter(mode="webhook", secret_token="secret", user_name="mybot") + clock.install(adapter) + chat = MagicMock() + chat.get_state.return_value = create_mock_state() + chat.process_message = MagicMock( + side_effect=lambda *_args, **_kwargs: asyncio.get_running_loop().create_task(asyncio.sleep(0)) + ) + adapter._chat = chat + + def part(message_id: int, **overrides: Any) -> dict[str, Any]: + message = _sample_message( + message_id=message_id, + chat={"id": -100, "type": "group"}, + media_group_id="g", + photo=[{"file_id": f"p{message_id}", "file_unique_id": f"u{message_id}", "width": 1, "height": 1}], + **overrides, + ) + message.pop("text") + return message + + # Telegram parsing fills neither ``links`` nor ``reply_to`` yet + # (#228), so the per-part values are injected after parsing. + quoted = [adapter.parse_telegram_message(_sample_message(message_id=n), "telegram:-100") for n in (90, 91)] + parse = adapter.parse_telegram_message + extras = {3: (quoted[0], "https://a.example"), 6: (quoted[1], "https://b.example")} + + def parse_with_extras(message: Any, thread_id: str) -> Message: + parsed = parse(message, thread_id) + if message["message_id"] in extras: + parsed.reply_to, url = extras[message["message_id"]] + parsed.links = [LinkPreview(url=url)] + return parsed + + adapter.parse_telegram_message = parse_with_extras # type: ignore[method-assign] + + # Eleven parts (the oldest falls out of the 10-part buffer), only an + # older part mentions the bot, and part 5 is redelivered. + parts = [part(n, **({"caption": "@mybot look"} if n == 2 else {})) for n in range(1, 12)] + [part(5)] + for index, message in enumerate(parts): + adapter.process_update({"update_id": index + 1, "message": message}) # type: ignore[typeddict-item] + await _flush() + await clock.advance(1_000) + + chat.process_message.assert_called_once() + parsed = chat.process_message.call_args.args[2] + assert [attachment.fetch_metadata["fileId"] for attachment in parsed.attachments] == [ + f"p{n}" for n in range(2, 12) + ] + assert parsed.id == "-100:11" + assert parsed.text == "@mybot look" + assert parsed.is_mention is True + assert [link.url for link in parsed.links] == ["https://a.example", "https://b.example"] + assert parsed.reply_to is quoted[0] + + @pytest.mark.asyncio + async def test_waituntil_settles_when_disconnect_cancels_a_settling_album(self): + """Python-specific: ``disconnect()`` cancels the album; the ``wait_until`` task still settles.""" + clock = _FakeClock() + adapter = _make_adapter(mode="webhook", secret_token="secret", user_name="mybot") + clock.install(adapter) + chat = MagicMock() + chat.get_state.return_value = create_mock_state() + adapter._chat = chat + waited: list[Any] = [] + message = _sample_message(chat={"id": -100, "type": "group"}, media_group_id="g", caption="c") + message.pop("text") + + [task] = adapter.process_update( + {"update_id": 1, "message": message}, # type: ignore[typeddict-item] + WebhookOptions(wait_until=waited.append), + ) + await _flush() + await adapter.disconnect() + + assert task.cancelled() + assert await waited[0] is None + chat.process_message.assert_not_called() + + +class TestTelegramPollingAcknowledgement: + """Polling waits for handlers before advancing ``offset`` (#942).""" + + @pytest.mark.asyncio + async def test_starts_polling_advances_offset_and_stops_cleanly(self): + fixture = _PollingFixture([_polled_message(10, chat_id=123, text="polled message")]) + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.chat.initialize() + await fixture.adapter.start_polling( + TelegramLongPollingConfig(limit=1, timeout=1, allowed_updates=["message"], retry_delay_ms=0) + ) + await _wait_until(lambda: handler.await_count > 0 and len(fixture.api.polls) >= 2) + await fixture.adapter.stop_polling() + + assert fixture.api.calls[1][0] == "deleteWebhook" + assert len(fixture.api.polls) == 2 + first, second = fixture.api.polls + assert first["limit"] == 1 + assert first["timeout"] == 1 + assert first["allowed_updates"] == ["message"] + assert "offset" not in first + assert second["offset"] == 11 + assert handler.await_count == 1 + assert handler.await_args.args[0].id == "telegram:123" + assert fixture.adapter.is_polling is False + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_waits_for_polled_message_processing_and_saves_failures_before_acknowledging_updates(self): + fixture = _PollingFixture([_polled_message(10, chat_id=123, text="polled message")]) + processing: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + async def handler(*_args: Any) -> None: + await processing + + fixture.chat.on_mention(handler) + try: + await fixture.start(limit=1, retry_delay_ms=0) + await _wait_until(lambda: len(fixture.api.polls) == 1) + await _flush() + assert len(fixture.api.polls) == 1 + + processing.set_exception(RuntimeError("Database admission failed")) + await _wait_until(lambda: len(fixture.api.polls) >= 2) + + assert fixture.api.polls[1]["offset"] == 11 + saved = await fixture.checkpoint() + assert saved["offset"] == 11 + [entry] = saved["pending"] + assert entry["update"]["update_id"] == 10 + assert entry["attempts"] == 1 + assert entry["retryAt"] == fixture.clock.now + 1_000 + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_coalesces_polled_media_groups_before_acknowledging_updates(self): + fixture = _PollingFixture(_polled_album(chat_id=123, first_id=10)) + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.start(limit=2, retry_delay_ms=0) + await _wait_until(lambda: len(fixture.api.polls) >= 2) + assert handler.await_count == 0 + + await fixture.clock.advance(2_100) + + assert handler.await_count == 1 + assert len(handler.await_args.args[1].attachments) == 2 + assert fixture.api.polls[1]["offset"] == 12 + assert await fixture.checkpoint() is None + # Cutting the collection poll short is not a request failure. + assert not [ + call for call in fixture.adapter._logger.warn.calls if call[0] == "Telegram polling request failed" + ] # type: ignore[attr-defined] + finally: + await fixture.stop() + + @pytest.mark.parametrize("kind", ["message", "command", "action", "reaction"]) + @pytest.mark.asyncio + async def test_saves_a_failed_update_before_acknowledgement_and_retains_it_until_retry_succeeds(self, kind: str): + update = _polled_kind(kind) + fixture = _PollingFixture([update]) + calls: list[Any] = [] + gate: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + async def handler(*args: Any) -> None: + calls.append(args) + if len(calls) == 1: + raise RuntimeError("Admission failed") + await gate + + fixture.chat.on_mention(handler) + fixture.chat.on_slash_command(handler) + fixture.chat.on_action(handler) + fixture.chat.on_reaction(handler) + try: + await fixture.start() + await _wait_until(lambda: len(calls) == 1 and len(fixture.api.polls) >= 2) + assert fixture.api.polls[-1]["offset"] == 2 + saved = await fixture.checkpoint() + assert saved["offset"] == 2 + assert [entry["update"] for entry in saved["pending"]] == [update] + + # The retry is due after 1 s; the loop polls once more first. + await fixture.clock.advance(1_100) + await _wait_until(lambda: len(calls) == 2) + assert (await fixture.checkpoint())["pending"] # kept until the retry settles + gate.set_result(None) + await _wait_until(lambda: _is_none(fixture.checkpoint())) + assert len(calls) == 2 + finally: + if not gate.done(): + gate.set_result(None) + await fixture.stop() + + @pytest.mark.asyncio + async def test_recovers_a_saved_failed_update_after_process_loss_despite_core_dedupe(self): + first = _PollingFixture([_polled_message(1)]) + first.chat.on_mention(AsyncMock(side_effect=RuntimeError("Admission failed"))) + try: + await first.start() + await _wait_until(lambda: len(first.api.polls) >= 2) + saved = await first.checkpoint() + assert saved["offset"] == 2 + finally: + await first.stop() + + state = MemoryStateAdapter() + await state.connect() + await state.set(_POLLING_CHECKPOINT, json.loads(json.dumps(saved))) + # The first run's core dedupe claim: a polled retry must bypass it. + await state.set("dedupe:telegram:1:1", True) + second = _PollingFixture([], state=state, clock=first.clock) + handler = AsyncMock() + second.chat.on_mention(handler) + try: + await second.start() + await second.clock.advance(1_100) + await _wait_until(lambda: _is_none(second.checkpoint())) + handler.assert_awaited_once() + assert second.api.polls + assert all(poll["offset"] == 2 for poll in second.api.polls) + finally: + await second.stop() + + @pytest.mark.asyncio + async def test_does_not_acknowledge_failed_ordinary_updates_when_saving_their_retry_fails(self): + ordinary = _polled_message(1) + fixture = _PollingFixture([ordinary, _polled_message(2, chat_id=2)]) + save = fixture.state.set + failing = True + + async def set_checkpoint(key: str, value: Any, ttl_ms: int | None = None) -> None: + if key == _POLLING_CHECKPOINT and failing: + raise RuntimeError("Storage unavailable") + await save(key, value, ttl_ms) + + fixture.state.set = set_checkpoint # type: ignore[method-assign] + handled: list[str] = [] + + async def handler(thread: Any, *_args: Any) -> None: + handled.append(thread.id) + if thread.id == "telegram:1": + raise RuntimeError("Reply failed") + + fixture.chat.on_mention(handler) + try: + await fixture.start() + await _wait_until(lambda: len(handled) == 2) + # The failed save backs off (10 ms) and polls again unacknowledged. + await fixture.clock.advance(10) + await _wait_until(lambda: len(fixture.api.polls) > 1) + assert all("offset" not in poll for poll in fixture.api.polls) + # The redelivered successful update is not handled twice. + assert handled.count("telegram:2") == 1 + + failing = False + await fixture.clock.advance(20) + await _wait_until(lambda: fixture.api.polls[-1].get("offset") == 3) + saved = await fixture.checkpoint() + assert saved["offset"] == 3 + assert [entry["update"] for entry in saved["pending"]] == [ordinary] + assert handled.count("telegram:2") == 1 + finally: + failing = False + await fixture.stop() + + @pytest.mark.asyncio + async def test_retries_every_member_of_a_failed_album(self): + fixture = _PollingFixture(_polled_album()) + handler = AsyncMock(side_effect=[RuntimeError("Admission failed"), None]) + fixture.chat.on_mention(handler) + try: + await fixture.start() + await fixture.clock.advance(2_100) + assert handler.await_count == 1 + saved = await fixture.checkpoint() + assert saved["offset"] == 3 + assert [(entry["update"]["update_id"], entry["attempts"]) for entry in saved["pending"]] == [(1, 1), (2, 1)] + + await fixture.clock.advance(3_000) + assert handler.await_count == 2 + for call in handler.await_args_list: + assert len(call.args[1].attachments) == 2 + assert await fixture.checkpoint() is None + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_keeps_pending_albums_through_cleanup_failures_without_repeating_successful_handlers(self): + fixture = _PollingFixture(_polled_album()) + remove = fixture.state.delete + failures = 0 + + async def delete_checkpoint(key: str) -> None: + nonlocal failures + if key == _POLLING_CHECKPOINT: + failures += 1 + if failures == 1: + raise RuntimeError("Cleanup failed") + await remove(key) + + fixture.state.delete = delete_checkpoint # type: ignore[method-assign] + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.start(limit=1) + await fixture.clock.advance(5_000) + handler.assert_awaited_once() + assert failures == 2 + assert await fixture.checkpoint() is None + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_does_not_buffer_albums_from_users_outside_the_allowlist(self): + fixture = _PollingFixture(_polled_album(), allowed_user_ids=["3"]) + written: list[str] = [] + save = fixture.state.set + + async def record(key: str, value: Any, ttl_ms: int | None = None) -> None: + written.append(key) + await save(key, value, ttl_ms) + + fixture.state.set = record # type: ignore[method-assign] + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.start(limit=1) + await _wait_until(lambda: fixture.api.polls and fixture.api.polls[-1].get("offset") == 3) + handler.assert_not_awaited() + assert _POLLING_CHECKPOINT not in written + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_polls_between_album_batches_when_another_album_becomes_ready_during_a_handler(self): + clock = _FakeClock() + state = MemoryStateAdapter() + await state.connect() + other_album = [ + { + **update, + "update_id": update["update_id"] + 2, + "message": {**update["message"], "chat": {"id": 2, "type": "private"}}, + } + for update in _polled_album() + ] + await state.set( + _POLLING_CHECKPOINT, + { + "offset": 5, + "pending": [{"update": update, "receivedAt": 0} for update in _polled_album()] + + [{"update": update, "receivedAt": clock.now} for update in other_album], + }, + ) + fixture = _PollingFixture([], state=state, clock=clock) + gate: asyncio.Future[None] = asyncio.get_running_loop().create_future() + order: list[str] = [] + + async def handler(thread: Any, *_args: Any) -> None: + order.append(thread.id) + if thread.id == "telegram:1": + await gate + raise RuntimeError("Reply failed") + + fixture.chat.on_mention(handler) + try: + await fixture.start() + await clock.advance(1_000) + await _wait_until(lambda: order == ["telegram:1"]) + # The other album is due now, but the loop must poll first. + fixture.api.deliver(_polled_message(5, chat_id=3)) + gate.set_result(None) + await _wait_until(lambda: "telegram:3" in order) + # The other album is already due, so the poll does not wait. + assert fixture.api.polls[0]["timeout"] == 0 + await clock.advance(1_000) + await _wait_until(lambda: "telegram:2" in order) + assert order[:3] == ["telegram:1", "telegram:3", "telegram:2"] + finally: + if not gate.done(): + gate.set_result(None) + await fixture.stop() + + @pytest.mark.parametrize(("attempts", "delay"), [(1, 2_000), (8, 30_000)]) + @pytest.mark.asyncio + async def test_retains_the_retry_count_and_bounds_album_backoff(self, attempts: int, delay: int): + state = MemoryStateAdapter() + await state.connect() + await state.set( + _POLLING_CHECKPOINT, + { + "offset": 3, + "pending": [ + {"update": update, "receivedAt": 0, "attempts": attempts, "retryAt": 0} + for update in _polled_album() + ], + }, + ) + fixture = _PollingFixture([], state=state) + fixture.chat.on_mention(AsyncMock(side_effect=RuntimeError("Reply failed"))) + try: + await fixture.start(retry_delay_ms=0) + await fixture.clock.advance(1_000) + saved = await fixture.checkpoint() + assert [entry["attempts"] for entry in saved["pending"]] == [attempts + 1, attempts + 1] + failed_at = fixture.clock.now + assert {entry["retryAt"] for entry in saved["pending"]} == {failed_at + delay} + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_rate_limited_update_waits_at_least_retry_after(self): + fixture = _PollingFixture([_polled_message(1)]) + fixture.chat.on_mention(AsyncMock(side_effect=AdapterRateLimitError("telegram", 60))) + try: + await fixture.start(retry_delay_ms=0) + await _wait_until(lambda: len(fixture.api.polls) >= 2) + saved = await fixture.checkpoint() + [entry] = saved["pending"] + assert entry["retryAt"] == fixture.clock.now + 60_000 + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_does_not_replay_another_bots_saved_album(self): + state = MemoryStateAdapter() + await state.connect() + saved = {"offset": 3, "pending": [{"update": update, "receivedAt": 0} for update in _polled_album()]} + await state.set(_POLLING_CHECKPOINT, saved) + fixture = _PollingFixture([], state=state, bot_id=1000) + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.start() + await _wait_until(lambda: len(fixture.api.polls) == 1) + await fixture.clock.advance(2_000) + handler.assert_not_awaited() + assert await fixture.checkpoint() == saved + assert "offset" not in fixture.api.polls[0] + finally: + await fixture.stop() + + @pytest.mark.parametrize("kind", ["message", "command", "action", "reaction"]) + @pytest.mark.asyncio + async def test_processes_beyond_a_full_page_despite_a_permanently_failing_update(self, kind: str): + update = _polled_kind(kind) + others = [_polled_message(update_id, chat_id=2) for update_id in range(2, 102)] + album = [ + { + **part, + "update_id": part["update_id"] + 101, + "message": {**part["message"], "chat": {"id": 3, "type": "private"}}, + } + for part in _polled_album() + ] + fixture = _PollingFixture([update, *others, *album]) + received: list[str] = [] + + async def fail(*_args: Any) -> None: + raise RuntimeError("Forbidden: bot was blocked by the user") + + async def on_mention(thread: Any, *_args: Any) -> None: + if thread.id == "telegram:1": + await fail() + received.append(thread.id) + + fixture.chat.on_mention(on_mention) + fixture.chat.on_slash_command(fail) + fixture.chat.on_action(fail) + fixture.chat.on_reaction(fail) + try: + await fixture.start(limit=100, retry_delay_ms=0) + await _wait_until(lambda: fixture.api.polls and fixture.api.polls[-1].get("offset") == 104, rounds=20_000) + assert all(poll["limit"] == 100 for poll in fixture.api.polls) + await fixture.clock.advance(2_100) + await _wait_until(lambda: "telegram:3" in received, rounds=20_000) + assert received.count("telegram:2") == 100 + saved = await fixture.checkpoint() + assert saved["offset"] == 104 + assert [entry["update"] for entry in saved["pending"]] == [update] + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_combines_an_album_spanning_a_full_polling_response(self): + parts = _polled_album() + updates = [ + { + **parts[index % 2], + "update_id": index + 1, + "message": {**parts[index % 2]["message"], "message_id": index + 1, "media_group_id": str(index // 3)}, + } + for index in range(102) + ] + fixture = _PollingFixture(updates) + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.start(limit=100) + # The last album straddles the page: its first part is in the + # first response and the other two in the second. + await _wait_until(lambda: len(fixture.api.polls) >= 3, rounds=20_000) + assert [poll.get("offset") for poll in fixture.api.polls[:3]] == [None, 101, 103] + for _ in range(10): + if handler.await_count == 34: + break + await fixture.clock.advance(1_000) + assert handler.await_count == 34 + for call in handler.await_args_list: + assert len(call.args[1].attachments) == 3 + await _wait_until(lambda: _is_none(fixture.checkpoint())) + finally: + await fixture.stop() + + @pytest.mark.asyncio + async def test_recovers_bot_identity_while_polling_after_a_failed_startup_get_me(self): + fixture = _PollingFixture([_polled_message(1)]) + api = fixture.api + get_me_calls = 0 + + async def flaky_get_me(method: str, payload: Any = None, **kwargs: Any) -> Any: + nonlocal get_me_calls + if method == "getMe": + get_me_calls += 1 + if get_me_calls == 1: + raise NetworkError("telegram", "getMe unavailable") + return await api(method, payload, **kwargs) + + fixture.adapter.telegram_fetch = flaky_get_me # type: ignore[method-assign] + fixture.chat.on_mention(AsyncMock(side_effect=RuntimeError("Reply failed"))) + try: + await fixture.start() + await _wait_until(lambda: len(fixture.api.polls) >= 2) + assert get_me_calls == 2 + # The failure is saved under the recovered bot's scope. + saved = await fixture.checkpoint() + assert [entry["update"]["update_id"] for entry in saved["pending"]] == [1] + assert await fixture.state.get("telegram:polling:None") is None + finally: + await fixture.stop() + + +class TestTelegramPollingShutdownPythonEdges: + """Python-specific: ``stop_polling`` / ``Chat.shutdown`` against the checkpoint.""" + + @pytest.mark.asyncio + async def test_stop_during_album_settle_returns_promptly_and_a_restart_delivers_it_once(self): + first = _PollingFixture(_polled_album()) + first_handler = AsyncMock() + first.chat.on_mention(first_handler) + await first.start(limit=1) + await _wait_until(lambda: first.api.polls and first.api.polls[-1].get("offset") == 3) + await _flush() + + # The loop is parked in the shortened collection poll; nothing has + # advanced the fake clock, so only cancellation can end it. + await first.adapter.stop_polling() + assert first.adapter.is_polling is False + first_handler.assert_not_awaited() + saved = await first.checkpoint() + assert saved["offset"] == 3 + assert [entry["update"]["update_id"] for entry in saved["pending"]] == [1, 2] + await first.chat.shutdown() + + second = _PollingFixture([], state=first.state, clock=first.clock) + handler = AsyncMock() + second.chat.on_mention(handler) + try: + await second.start(limit=1) + await second.clock.advance(3_000) + handler.assert_awaited_once() + assert len(handler.await_args.args[1].attachments) == 2 + assert await second.checkpoint() is None + assert all(poll["offset"] == 3 for poll in second.api.polls) + finally: + await second.stop() + + @pytest.mark.asyncio + async def test_chat_shutdown_during_album_settle_does_not_wait_for_the_album(self): + fixture = _PollingFixture(_polled_album()) + handler = AsyncMock() + fixture.chat.on_mention(handler) + await fixture.start() + await _wait_until(lambda: len(fixture.api.polls) >= 2) + # Due album: the loop dispatches it, and the parts park in their + # settle sleep (the fake clock is not advanced again). + await fixture.clock.advance(1_050) + assert fixture.adapter._media_group_tasks + + await asyncio.wait_for(fixture.chat.shutdown(), timeout=5) + + handler.assert_not_awaited() + assert fixture.adapter.is_polling is False + saved = await fixture.checkpoint() + assert [(entry["update"]["update_id"], entry["attempts"]) for entry in saved["pending"]] == [(1, 1), (2, 1)] + + @pytest.mark.asyncio + async def test_stop_during_the_checkpoint_read_dispatches_no_saved_retry(self): + state = MemoryStateAdapter() + await state.connect() + saved = {"offset": 2, "pending": [{"update": _polled_message(1), "receivedAt": 0, "attempts": 1, "retryAt": 0}]} + await state.set(_POLLING_CHECKPOINT, saved) + fixture = _PollingFixture([], state=state) + read_started = asyncio.Event() + release: asyncio.Future[None] = asyncio.get_running_loop().create_future() + read = state.get + + async def slow_get(key: str) -> Any: + if key == _POLLING_CHECKPOINT: + read_started.set() + await release + return await read(key) + + state.get = slow_get # type: ignore[method-assign] + handler = AsyncMock() + fixture.chat.on_mention(handler) + try: + await fixture.start() + await read_started.wait() + stopping = asyncio.create_task(fixture.adapter.stop_polling()) + await _flush() + release.set_result(None) + await stopping + + handler.assert_not_awaited() + assert fixture.api.polls == [] + state.get = read # type: ignore[method-assign] + assert await fixture.checkpoint() == saved + finally: + if not release.done(): + release.set_result(None) + await fixture.stop() + + @pytest.mark.asyncio + async def test_stop_polling_waits_for_an_in_flight_handler(self): + fixture = _PollingFixture([_polled_message(1)]) + gate: asyncio.Future[None] = asyncio.get_running_loop().create_future() + finished: list[bool] = [] + + async def handler(*_args: Any) -> None: + await gate + finished.append(True) + + fixture.chat.on_mention(handler) + try: + await fixture.start() + await _wait_until(lambda: len(fixture.api.polls) == 1) + await _flush() + stopping = asyncio.create_task(fixture.adapter.stop_polling()) + await _flush() + assert not stopping.done() + + gate.set_result(None) + await stopping + assert finished == [True] + assert len(fixture.api.polls) == 1 + assert fixture.adapter.is_polling is False + finally: + if not gate.done(): + gate.set_result(None) + await fixture.stop() + + @pytest.mark.asyncio + async def test_chat_shutdown_cancelling_a_handler_keeps_the_update_pending(self): + fixture = _PollingFixture([_polled_message(1)]) + writes: list[Any] = [] + save = fixture.state.set + + async def record(key: str, value: Any, ttl_ms: int | None = None) -> None: + if key == _POLLING_CHECKPOINT: + writes.append(value) + await save(key, value, ttl_ms) + + fixture.state.set = record # type: ignore[method-assign] + never: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + async def handler(*_args: Any) -> None: + await never + + fixture.chat.on_mention(handler) + await fixture.start() + await _wait_until(lambda: len(fixture.api.polls) == 1) + await _flush() + + await fixture.chat.shutdown() + + assert fixture.adapter.is_polling is False + assert len(fixture.api.polls) <= 2 + [checkpoint] = writes + [entry] = checkpoint["pending"] + assert entry["update"]["update_id"] == 1 + assert entry["attempts"] == 1 + + +async def _is_none(value: Any) -> bool: + return (await value) is None