diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index 78ee097..96e09aa 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -624,6 +624,7 @@ stay explicit instead of being rediscovered in code review. | Teams native streaming `streamId` placement on `streaminfo` entity (DMs) | `streamId` is added to BOTH `channelData` and the `streaminfo` entity on subsequent and final activities (it is omitted from both on the first chunk because the server has not yet assigned an id) | `IStreamer.emit` handles the wire shape; the upstream JS SDK includes `streamId` on both sites as part of its `streaminfo` entity contract per [MS Bot Framework streaming docs](https://learn.microsoft.com/microsoftteams/platform/bots/streaming-ux#continue-streaming) | Bot Framework REST contract requires the entity-level `streamId` for continuation activities — earlier Python versions set it only on `channelData`, which Teams may treat as a malformed continuation and detach from the original stream. This row documents the contract, not a divergence; both Python and the JS SDK comply. | | Teams native streaming final-send when first chunk's `id` was empty (DMs) | `_close_stream_session` sends the final `message` activity whenever `text` is non-empty, even if `stream_id` is `None` (Bot Framework REST response returned `{"id": ""}` on the first chunk). The final activity omits `streamId` from `channelData` rather than serializing `None`. | Upstream's `streamViaEmit` awaits the `chunk` event for the first activity's `id`; if Teams returns an empty id, `messageId` becomes `""` and the SDK's auto-close emits the final activity through `IStreamer` regardless | Without the final `message` activity, the Teams client's streaming UI keeps spinning until the platform times the session out — a stuck-loading-state UX failure with no user workaround. We mirror upstream's looser check (text non-empty → ship the final) so the streaming indicator clears even when the Bot Framework REST response surface returns an empty `id`. | | `RawMessage.text` override field — **transitional** | New optional `text: str \| None = None` field on `RawMessage` (`src/chat_sdk/types.py`). When set by an adapter, `Thread._handle_stream` MUST prefer it over its own local accumulator when constructing the recorded `SentMessage` body / message-history entry. `None` falls back to the local accumulator (backward-compatible default for adapters that don't need the override). Set by `_stream_via_emit` so the recorded message matches what Teams actually shipped, even when chunks were buffered into the throttle window and cancellation skipped the flush emit. | Upstream's `RawMessage` (`packages/chat/src/types.ts`) is `{ id; raw; threadId }` only. Cancellation-text reconciliation lives inside `@microsoft/teams.apps`'s `IStreamer.emit` (the npm SDK owns the buffer and never surfaces a buffered-but-unsent suffix to `chat`). | Direct consequence of the hand-rolled Teams native streaming row above. Without the override, the SDK's local accumulator (which captures every chunk yielded to the adapter, including chunks coalesced into the throttle window) would diverge from what Teams actually accepted whenever a session is canceled with buffered text pending — recording text the user never saw. Disappears alongside the hand-rolled wire format once we migrate to `microsoft-teams-apps` (Python). Regression coverage: `tests/test_thread_faithful.py::test_should_prefer_raw_message_text_override_over_local_accumulator` (would fail if someone "fixes" Thread.stream back to upstream's local-accumulator-only behavior) and `tests/test_teams_native_streaming.py::test_canceled_stream_sets_raw_message_text_override`. | +| `Thread._handle_stream` does NOT seed `StreamOptions.update_interval_ms` (vercel/chat#340) | The `StreamOptions` handed to `adapter.stream` (and the fallback) leaves `update_interval_ms = None` unless the caller (`StreamingPlan`) supplied one. Adapters that consume the field apply their own default when it is `None`: Telegram → `TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS` (250ms); `_fallback_stream` → `self._streaming_update_interval_ms`. | Upstream `thread.ts` seeds `updateIntervalMs: this._streamingUpdateIntervalMs` (500ms default) into the options object before spreading caller options, so the adapter always sees a concrete interval (upstream's `thread.test.ts > should fall back when adapter.stream returns null` asserts `objectContaining({ updateIntervalMs: 500 })`). | Our hand-rolled Teams native streaming path (the transitional rows above) treats any non-`None` `update_interval_ms` as a **caller override** of its 1500ms quota-protecting emit throttle. Seeding the 500ms thread default would silently drop Teams DM throttling below Teams' ~1 req/sec streaming-endpoint quota. Upstream's Teams adapter delegates throttling to `@microsoft/teams.apps` and ignores the field, so the seed is harmless there; ours can't, so we keep the field unseeded until the `microsoft-teams-apps` migration (issue #93) lands and `TeamsAdapter` stops owning the throttle. Observable parity for the fallback path is preserved because `_fallback_stream` already substitutes `self._streaming_update_interval_ms` when the field is `None`. Regression coverage: `tests/test_thread_faithful.py::test_should_fall_back_when_adapterstream_returns_null` (asserts `update_interval_ms is None`, with a comment pointing back here) and `tests/test_telegram_streaming.py::test_stream_defaults_update_interval_when_options_omitted`. Reconcile `TeamsAdapter._resolve_emit_interval` first if this seed is ever added. | | Teams divider rendering | `card_to_adaptive_card` hoists `separator: True` onto the next sibling (or emits a non-empty Container for a trailing divider) | `convertDividerToElement` emits an empty `Container` with `separator: True` | Upstream shares the same bug: Microsoft Teams renders an empty Container at zero height, so the separator line is effectively invisible. Python port fixes locally (issue #45) rather than blocking on upstream. | | `SlackAdapter.current_token` / `current_token_async` / `current_client` | Public accessors that return the request-context-bound token and a preconfigured `AsyncWebClient`. `current_token` (sync `@property`) reads the cache; `current_token_async` (async method) invokes the resolver on demand for callable `bot_token` configs used outside `handle_webhook`. | Not exposed (`getToken()` is private on the TS `SlackAdapter`) | Python-only addition (issue #47). Downstream code that calls Slack Web APIs from inside a handler — email resolution, user profile fetches, reaction bookkeeping — otherwise depends on underscore-prefixed helpers. The async variant is required because the sync `current_token` cannot drive an async resolver (see `bot_token` resolver invocation site row). | | `SlackAdapterConfig.webhook_verifier` | Optional `Callable[[request, body], bool \| str \| None \| Awaitable[...]]` that fully replaces signing-secret HMAC verification. Lets callers integrate platform-managed verification (e.g. Slack Enterprise Grid edge proxies, KMS-signed payloads, test harness escape hatches). `webhook_verifier` takes precedence over both `signing_secret` (config) and the `SLACK_SIGNING_SECRET` env var — when set, both are ignored. | Upstream has its own `webhookVerifier` field on `SlackAdapterConfig` and matches this precedence direction after vercel/chat#468 (commit `0f0c203`, chat@4.29.0). | Behavior parity restored in 0.4.29 sync wave. The original Python port (PR #87, 0.4.27) preferred `signing_secret` to match upstream's intent at that time; upstream reversed itself in #468 so an env-configured `SLACK_SIGNING_SECRET` could not silently shadow a verifier the caller wired up. This port follows. The contract is documented as a SECURITY surface in `slack/types.py` (`SlackWebhookVerifier`): returning truthy passes the request, falsy/None rejects 401, and a `str` substitutes the request body before dispatch. | diff --git a/src/chat_sdk/adapters/telegram/adapter.py b/src/chat_sdk/adapters/telegram/adapter.py index 92efff7..8e03a2c 100644 --- a/src/chat_sdk/adapters/telegram/adapter.py +++ b/src/chat_sdk/adapters/telegram/adapter.py @@ -17,9 +17,11 @@ import math import os import re +import time +from collections.abc import AsyncIterable, Awaitable, Callable from dataclasses import dataclass from datetime import datetime, timezone -from typing import Any, cast +from typing import Any, TypeVar, cast from chat_sdk.adapters.telegram.cards import ( card_to_telegram_inline_keyboard, @@ -62,6 +64,8 @@ ResourceNotFoundError, ValidationError, ) +from chat_sdk.shared.markdown_parser import ast_to_plain_text, parse_markdown +from chat_sdk.shared.streaming_markdown import StreamingMarkdownRenderer from chat_sdk.types import ( ActionEvent, AdapterPostableMessage, @@ -76,8 +80,11 @@ LockScope, Message, MessageMetadata, + PostableMarkdown, RawMessage, ReactionEvent, + StreamChunk, + StreamOptions, ThreadInfo, UserInfo, WebhookOptions, @@ -109,6 +116,12 @@ TELEGRAM_DEFAULT_POLLING_TIMEOUT_SECONDS = 30 TELEGRAM_DEFAULT_POLLING_LIMIT = 100 TELEGRAM_DEFAULT_POLLING_RETRY_DELAY_MS = 1000 +TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS = 250 +# 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 +# /can't parse (?:caption )?entities/i test(). +TELEGRAM_MARKDOWN_PARSE_ERROR_PATTERN = re.compile(r"can't parse (?:caption )?entities", re.IGNORECASE) TELEGRAM_MAX_POLLING_LIMIT = 100 TELEGRAM_MIN_POLLING_LIMIT = 1 TELEGRAM_MIN_POLLING_TIMEOUT_SECONDS = 0 @@ -144,12 +157,26 @@ class ResolvedTelegramLongPollingConfig: TelegramRuntimeMode = str # "webhook" | "polling" +_T = TypeVar("_T") + # --------------------------------------------------------------------------- # Module-level helpers # --------------------------------------------------------------------------- +def _markdown_to_plain_text(markdown: str) -> str: + """Extract plain text from a markdown string, stripping all formatting. + + Local equivalent of upstream's ``markdownToPlainText`` chat export + (``parseMarkdown`` + ``mdast-util-to-string``). Unclosed inline markers + (``**broken``) stay literal text in our parser, so they survive into + the plain-text rendering — matching remark, which also treats them as + literal when no closer exists. + """ + return ast_to_plain_text(parse_markdown(markdown)) + + def _utf16_len(text: str) -> int: """Return the length of *text* measured in UTF-16 code units.""" return len(text.encode("utf-16-le")) // 2 @@ -620,6 +647,12 @@ def __init__(self, config: TelegramAdapterConfig | None = None) -> None: self._polling_task: asyncio.Task[None] | None = None self._polling_active: bool = False + # Draft-id counter for native DM draft streaming (vercel/chat#340). + # Seeded from wall-clock millis (mod int32 max) so concurrent bot + # restarts don't reuse the previous process's draft ids; wraps to 1 + # at 2_147_483_647 to stay within Telegram's signed-int32 range. + self._next_draft_id: int = max(1, int(time.time() * 1000) % 2_147_483_647) + # Shared aiohttp session for connection pooling self._http_session: Any | None = None @@ -1200,6 +1233,12 @@ async def post_message( card = extract_card(message) reply_markup = card_to_telegram_inline_keyboard(card) if card else None parse_mode = self.resolve_parse_mode(message, card) + # Plain-text rendering of the same message, used as the retry body + # when Telegram rejects the MarkdownV2 entities (vercel/chat#340). + plain_text = self.truncate_message( + convert_emoji_placeholders(self.render_plain_text_message(message, card), "gchat"), + None, + ) text = self.truncate_message( convert_emoji_placeholders( # Route the card's standard-markdown fallback through the @@ -1239,7 +1278,7 @@ async def post_message( file = files[0] if not file: raise ValidationError("telegram", "File upload payload is empty") - raw_message = await self.send_document(parsed_thread, file, text, reply_markup, parse_mode) + raw_message = await self.send_document(parsed_thread, file, text, plain_text, reply_markup, parse_mode) elif len(attachments) == 1: attachment = attachments[0] if not attachment: @@ -1248,6 +1287,7 @@ async def post_message( parsed_thread, attachment, text, + plain_text, reply_markup, parse_mode, ) @@ -1255,15 +1295,25 @@ async def post_message( if not text.strip(): raise ValidationError("telegram", "Message text cannot be empty") - raw_message = await self.telegram_fetch( - "sendMessage", - { - "chat_id": parsed_thread.chat_id, - "message_thread_id": parsed_thread.message_thread_id, - "text": text, - "reply_markup": reply_markup, - "parse_mode": parse_mode, - }, + async def _send_message(resolved_parse_mode: str | None, resolved_text: str) -> TelegramMessage: + return await self.telegram_fetch( + "sendMessage", + { + "chat_id": parsed_thread.chat_id, + "message_thread_id": parsed_thread.message_thread_id, + "text": resolved_text, + "reply_markup": reply_markup, + "parse_mode": resolved_parse_mode, + }, + ) + + raw_message = await self.with_telegram_markdown_fallback( + parse_mode, + _send_message, + initial_text=text, + fallback_text=plain_text, + method="sendMessage", + thread_id=thread_id, ) resulting_thread_id = self.encode_thread_id( @@ -1310,6 +1360,12 @@ async def edit_message( card = extract_card(message) reply_markup = card_to_telegram_inline_keyboard(card) if card else None parse_mode = self.resolve_parse_mode(message, card) + # Plain-text rendering of the same message, used as the retry body + # when Telegram rejects the MarkdownV2 entities (vercel/chat#340). + plain_text = self.truncate_message( + convert_emoji_placeholders(self.render_plain_text_message(message, card), "gchat"), + None, + ) text = self.truncate_message( convert_emoji_placeholders( self._format_converter.from_markdown(card_to_fallback_text(card)) @@ -1323,15 +1379,29 @@ async def edit_message( if not text.strip(): raise ValidationError("telegram", "Message text cannot be empty") - result = await self.telegram_fetch( - "editMessageText", - { - "chat_id": chat_id, - "message_id": telegram_message_id, - "text": text, - "reply_markup": reply_markup or empty_telegram_inline_keyboard(), - "parse_mode": parse_mode, - }, + # Returns ``Any`` because Telegram answers ``true`` (not a Message) + # when editing inline messages — the ``result is True`` narrowing + # below handles that shape, matching the pre-#340 untyped fetch. + async def _edit_message_text(resolved_parse_mode: str | None, resolved_text: str) -> Any: + return await self.telegram_fetch( + "editMessageText", + { + "chat_id": chat_id, + "message_id": telegram_message_id, + "text": resolved_text, + "reply_markup": reply_markup or empty_telegram_inline_keyboard(), + "parse_mode": resolved_parse_mode, + }, + ) + + result = await self.with_telegram_markdown_fallback( + parse_mode, + _edit_message_text, + initial_text=text, + fallback_text=plain_text, + message_id=message_id, + method="editMessageText", + thread_id=thread_id, ) # Telegram returns ``true`` when editing inline messages @@ -1456,6 +1526,171 @@ async def start_typing(self, thread_id: str, status: str | None = None) -> None: }, ) + # -- Streaming ------------------------------------------------------------- + + async def stream( + self, + thread_id: str, + text_stream: AsyncIterable[str | StreamChunk], + options: StreamOptions | None = None, + ) -> RawMessage | None: + """Stream a message to a Telegram private chat via draft updates. + + Port of upstream ``TelegramAdapter.stream`` (vercel/chat#340). + Private chats (DMs) get native draft streaming through the + ``sendMessageDraft`` Bot API method: the draft bubble updates in + place as chunks arrive, throttled to ``options.update_interval_ms`` + (default :data:`TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS`), and a + regular ``sendMessage`` persists the final text when the stream + ends. Returns ``None`` for non-DM threads — before consuming any + chunks — so the SDK's built-in post+edit fallback handles groups, + supergroups, and channels. + + Draft updates render the in-flight markdown through + :class:`StreamingMarkdownRenderer` and + :func:`truncate_for_telegram`, so transiently unpaired entity + markers are trimmed to a MarkdownV2-safe boundary instead of + tripping Telegram's ``can't parse entities`` 400. If Telegram still + rejects the markdown, the stream downgrades to plain-text drafts + (and a plain-text final send); any other draft failure disables + draft updates entirely but never fails the stream — the final + message is always attempted. + """ + if not self.is_dm(thread_id): + return None + + parsed_thread = self._resolve_thread_id(thread_id) + update_interval_ms = self.clamp_integer( + options.update_interval_ms if options is not None else None, + TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS, + 0, + 2**53 - 1, + ) + + renderer = StreamingMarkdownRenderer() + draft_id = self.create_draft_id() + accumulated = "" + last_draft_text: str | None = None + last_flush_at = 0.0 + draft_streaming_enabled = True + stream_uses_markdown = True + + def _now_ms() -> float: + return time.monotonic() * 1000.0 + + def render_markdown_text(text: str) -> str: + return self.truncate_message( + convert_emoji_placeholders(self._format_converter.from_markdown(text), "gchat"), + TELEGRAM_MARKDOWN_PARSE_MODE, + ) + + def render_plain_text(text: str) -> str: + return self.truncate_message( + self.resolve_telegram_fallback_text(text, _markdown_to_plain_text(text)), + None, + ) + + def _draft_payload(text: str, *, markdown: bool) -> dict[str, Any]: + payload: dict[str, Any] = { + "chat_id": parsed_thread.chat_id, + "draft_id": draft_id, + "text": text, + } + # Omit absent optional keys: DMs have no forum-topic thread id, + # and plain-text drafts ship without a parse_mode (mirrors + # upstream, where JSON.stringify drops the undefined fields). + if parsed_thread.message_thread_id is not None: + payload["message_thread_id"] = parsed_thread.message_thread_id + if markdown: + payload["parse_mode"] = TELEGRAM_MARKDOWN_PARSE_MODE + return payload + + async def send_draft(text: str, use_markdown: bool) -> None: + nonlocal draft_streaming_enabled, last_draft_text, last_flush_at, stream_uses_markdown + if not draft_streaming_enabled or text == last_draft_text: + return + + try: + await self.telegram_fetch("sendMessageDraft", _draft_payload(text, markdown=use_markdown)) + last_draft_text = text + last_flush_at = _now_ms() + except Exception as error: + if use_markdown and self.is_telegram_markdown_parse_error(error): + # Telegram rejected the MarkdownV2 entities: downgrade + # this stream to plain-text drafts and retry once with + # the plain rendering of everything accumulated so far. + stream_uses_markdown = False + plain_draft_text = render_plain_text(accumulated) + try: + await self.telegram_fetch("sendMessageDraft", _draft_payload(plain_draft_text, markdown=False)) + last_draft_text = plain_draft_text + last_flush_at = _now_ms() + except Exception as retry_error: + draft_streaming_enabled = False + self._logger.warn( + "Telegram draft streaming update failed", + {"error": str(retry_error), "thread_id": thread_id}, + ) + return + + draft_streaming_enabled = False + self._logger.warn( + "Telegram draft streaming update failed", + {"error": str(error), "thread_id": thread_id}, + ) + + async def flush_draft() -> None: + if not draft_streaming_enabled: + return + draft_text = ( + render_markdown_text(renderer.render()) if stream_uses_markdown else render_plain_text(accumulated) + ) + await send_draft(draft_text, stream_uses_markdown) + + # Open the draft bubble immediately so the user sees activity + # before the first chunk lands. + await send_draft("", False) + + async for chunk in text_stream: + text: str | None = None + if isinstance(chunk, str): + text = chunk + elif isinstance(chunk, dict) and chunk.get("type") == "markdown_text": + text = chunk.get("text", "") + elif hasattr(chunk, "type") and getattr(chunk, "type", None) == "markdown_text": + # Runtime-narrowed to a MarkdownTextChunk via the `type` + # tag; only that variant has `.text`. Pyrefly doesn't do + # tag-based union narrowing, so read via `getattr`. + text = getattr(chunk, "text", "") + + if text is None: + # Task/plan progress chunks have no draft representation. + continue + + accumulated += text + renderer.push(text) + + if _now_ms() - last_flush_at >= update_interval_ms: + await flush_draft() + + await flush_draft() + + if not accumulated.strip(): + raise ValidationError("telegram", "Telegram streaming requires text content") + + # Persist the final message through the regular post path (which + # carries its own markdown-parse retry). The returned RawMessage + # leaves ``text`` unset so ``Thread.stream`` records its local + # accumulator — matching upstream, which records + # ``{ markdown: accumulated }``. + final_postable: AdapterPostableMessage = ( + PostableMarkdown(markdown=accumulated) + if stream_uses_markdown + else self.resolve_telegram_fallback_text(accumulated, _markdown_to_plain_text(accumulated)) + ) + + return await self.post_message(thread_id, final_postable) + # -- Fetching messages --------------------------------------------------- async def fetch_messages( @@ -1849,16 +2084,53 @@ async def send_document( thread: TelegramThreadId, file: Any, text: str, + plain_text: str, reply_markup: TelegramInlineKeyboardMarkup | None = None, parse_mode: str | None = None, ) -> TelegramMessage: """Send a document (file upload) to Telegram.""" - import aiohttp - data = getattr(file, "data", b"") if isinstance(data, memoryview) or not isinstance(data, bytes): data = bytes(data) + async def _send(resolved_parse_mode: str | None, resolved_text: str) -> TelegramMessage: + # FormData is rebuilt per attempt: aiohttp marks a FormData + # instance processed after one send, so the markdown-parse + # retry needs a fresh one (upstream rebuilds for the same + # reason — undici FormData bodies are single-use streams). + return await self.telegram_fetch( + "sendDocument", + self._create_telegram_document_form_data( + thread, + file, + data, + resolved_text, + reply_markup, + resolved_parse_mode, + ), + ) + + return await self.with_telegram_markdown_fallback( + parse_mode, + _send, + initial_text=text, + fallback_text=plain_text, + method="sendDocument", + thread_id=self.encode_thread_id(thread), + ) + + def _create_telegram_document_form_data( + self, + thread: TelegramThreadId, + file: Any, + data: bytes, + text: str, + reply_markup: TelegramInlineKeyboardMarkup | None = None, + parse_mode: str | None = None, + ) -> Any: + """Build the multipart form body for a ``sendDocument`` call.""" + import aiohttp + form_data = aiohttp.FormData() form_data.add_field("chat_id", thread.chat_id) @@ -1882,13 +2154,14 @@ async def send_document( if reply_markup: form_data.add_field("reply_markup", json.dumps(reply_markup)) - return await self.telegram_fetch("sendDocument", form_data) + return form_data async def send_attachment( self, thread: TelegramThreadId, attachment: Attachment, text: str, + plain_text: str, reply_markup: TelegramInlineKeyboardMarkup | None = None, parse_mode: str | None = None, ) -> TelegramMessage: @@ -1919,62 +2192,73 @@ async def send_attachment( f"Attachment data or URL required for {attachment.type}", ) - if data is None: - payload: dict[str, Any] = { - "chat_id": thread.chat_id, - upload["field"]: attachment.url, - } + if data is not None and (isinstance(data, memoryview) or not isinstance(data, bytes)): + data = bytes(data) - if isinstance(thread.message_thread_id, int): - payload["message_thread_id"] = thread.message_thread_id + async def _send(resolved_parse_mode: str | None, resolved_text: str) -> TelegramMessage: + if data is None: + payload: dict[str, Any] = { + "chat_id": thread.chat_id, + upload["field"]: attachment.url, + } - if text.strip(): - payload["caption"] = self.truncate_caption(text, parse_mode) - if parse_mode: - payload["parse_mode"] = parse_mode + if isinstance(thread.message_thread_id, int): + payload["message_thread_id"] = thread.message_thread_id - if attachment.type == "video": - if isinstance(attachment.width, int): - payload["width"] = attachment.width - if isinstance(attachment.height, int): - payload["height"] = attachment.height + if resolved_text.strip(): + payload["caption"] = self.truncate_caption(resolved_text, resolved_parse_mode) + if resolved_parse_mode: + payload["parse_mode"] = resolved_parse_mode - if reply_markup: - payload["reply_markup"] = reply_markup + if attachment.type == "video": + if isinstance(attachment.width, int): + payload["width"] = attachment.width + if isinstance(attachment.height, int): + payload["height"] = attachment.height - return await self.telegram_fetch(upload["method"], payload) + if reply_markup: + payload["reply_markup"] = reply_markup - if isinstance(data, memoryview) or not isinstance(data, bytes): - data = bytes(data) + return await self.telegram_fetch(upload["method"], payload) - form_data = aiohttp.FormData() - form_data.add_field("chat_id", thread.chat_id) + # FormData is rebuilt per attempt — see send_document. + form_data = aiohttp.FormData() + form_data.add_field("chat_id", thread.chat_id) - if isinstance(thread.message_thread_id, int): - form_data.add_field("message_thread_id", str(thread.message_thread_id)) + if isinstance(thread.message_thread_id, int): + form_data.add_field("message_thread_id", str(thread.message_thread_id)) - if text.strip(): - form_data.add_field("caption", self.truncate_caption(text, parse_mode)) - if parse_mode: - form_data.add_field("parse_mode", parse_mode) + if resolved_text.strip(): + form_data.add_field("caption", self.truncate_caption(resolved_text, resolved_parse_mode)) + if resolved_parse_mode: + form_data.add_field("parse_mode", resolved_parse_mode) - if attachment.type == "video": - if isinstance(attachment.width, int): - form_data.add_field("width", str(attachment.width)) - if isinstance(attachment.height, int): - form_data.add_field("height", str(attachment.height)) + if attachment.type == "video": + if isinstance(attachment.width, int): + form_data.add_field("width", str(attachment.width)) + if isinstance(attachment.height, int): + form_data.add_field("height", str(attachment.height)) - form_data.add_field( - upload["field"], - data, - filename=attachment.name if attachment.name is not None else "attachment", - content_type=attachment.mime_type if attachment.mime_type is not None else "application/octet-stream", - ) + form_data.add_field( + upload["field"], + data, + filename=attachment.name if attachment.name is not None else "attachment", + content_type=attachment.mime_type if attachment.mime_type is not None else "application/octet-stream", + ) - if reply_markup: - form_data.add_field("reply_markup", json.dumps(reply_markup)) + if reply_markup: + form_data.add_field("reply_markup", json.dumps(reply_markup)) - return await self.telegram_fetch(upload["method"], form_data) + return await self.telegram_fetch(upload["method"], form_data) + + return await self.with_telegram_markdown_fallback( + parse_mode, + _send, + initial_text=text, + fallback_text=plain_text, + method=upload["method"], + thread_id=self.encode_thread_id(thread), + ) # -- Message caching ----------------------------------------------------- @@ -2025,6 +2309,16 @@ def message_sequence(self, message_id: str) -> int: match = MESSAGE_SEQUENCE_PATTERN.search(message_id) return int(match.group(1)) if match else 0 + def create_draft_id(self) -> int: + """Return the next draft id for native DM draft streaming. + + Monotonically increasing per adapter instance so concurrent streams + to the same chat update distinct draft bubbles; wraps to 1 past + Telegram's signed-int32 maximum. + """ + self._next_draft_id = 1 if self._next_draft_id >= 2_147_483_647 else self._next_draft_id + 1 + return self._next_draft_id + # -- Pagination ---------------------------------------------------------- def paginate_messages( @@ -2239,6 +2533,51 @@ def resolve_parse_mode( # format_converter.render_postable, which emits MarkdownV2. return TELEGRAM_MARKDOWN_PARSE_MODE + def render_plain_text_message( + self, + message: AdapterPostableMessage, + card: Any, + ) -> str: + """Render a postable message as plain text (no MarkdownV2 markup). + + Port of upstream ``renderPlainTextMessage`` (vercel/chat#340): the + retry body used when Telegram rejects the MarkdownV2 entities of + the primary rendering. Handles both dict- and dataclass-shaped + postables (mirroring ``render_postable``'s branch order: raw → + markdown → ast). + """ + if card: + return card_to_fallback_text(card) + if isinstance(message, str): + return message + if isinstance(message, dict): + if "raw" in message: + return message["raw"] + if "markdown" in message: + return self.resolve_telegram_fallback_text( + message["markdown"], _markdown_to_plain_text(message["markdown"]) + ) + if "ast" in message: + return ast_to_plain_text(message["ast"]) + else: + if hasattr(message, "raw"): + return message.raw + if hasattr(message, "markdown"): + markdown = getattr(message, "markdown", "") + return self.resolve_telegram_fallback_text(markdown, _markdown_to_plain_text(markdown)) + if hasattr(message, "ast"): + return ast_to_plain_text(getattr(message, "ast", {})) + return self._format_converter.render_postable(message) + + def resolve_telegram_fallback_text(self, original_text: str, fallback_text: str) -> str: + """Prefer *fallback_text* unless stripping markup left it empty. + + A whitespace-only plain rendering (e.g. ``**`` — markers with no + content) falls back to *original_text* so the retry never sends a + body Telegram rejects as empty. + """ + return fallback_text if fallback_text.strip() else original_text + # -- Truncation ---------------------------------------------------------- def truncate_message(self, text: str, parse_mode: str | None = None) -> str: @@ -2413,6 +2752,67 @@ def throw_telegram_api_error( f"{description} (status {status}, error {error_code})", ) + # -- Markdown parse-error fallback (vercel/chat#340) ----------------------- + + async def with_telegram_markdown_fallback( + self, + parse_mode: str | None, + operation: Callable[[str | None, str], Awaitable[_T]], + *, + initial_text: str, + fallback_text: str, + method: str, + message_id: str | None = None, + thread_id: str | None = None, + ) -> _T: + """Run *operation*, retrying without ``parse_mode`` on entity errors. + + ``operation`` receives ``(parse_mode, text)`` and must build its own + request body per attempt (multipart bodies are single-use). The + first attempt ships ``(parse_mode, initial_text)``; when Telegram + rejects MarkdownV2 with a ``can't parse entities`` 400, the retry + ships ``(None, fallback)`` — the plain rendering, or the original + text when stripping markup left nothing (see + :meth:`resolve_telegram_fallback_text`). Every other failure, and + any failure for non-MarkdownV2 sends, propagates unchanged. + """ + try: + return await operation(parse_mode, initial_text) + except Exception as error: + if parse_mode != TELEGRAM_MARKDOWN_PARSE_MODE or not self.is_telegram_markdown_parse_error(error): + raise + + log_context: dict[str, Any] = { + "error": str(error), + "initial_text": initial_text, + "fallback_text": fallback_text, + "method": method, + } + if message_id is not None: + log_context["message_id"] = message_id + if thread_id is not None: + log_context["thread_id"] = thread_id + self._logger.warn( + "Telegram markdown parse failed; retrying without parse mode", + log_context, + ) + + return await operation(None, self.resolve_telegram_fallback_text(initial_text, fallback_text)) + + def is_telegram_markdown_parse_error(self, error: object) -> bool: + """Whether *error* is Telegram rejecting MarkdownV2 entity parsing. + + Matches the ``can't parse entities`` / ``can't parse caption + entities`` 400s that :meth:`throw_telegram_api_error` surfaces as + :class:`ValidationError`. Other 4xx validation failures (wrong + chat, empty text, …) do not qualify and must propagate. + """ + return ( + isinstance(error, ValidationError) + and error.adapter == "telegram" + and TELEGRAM_MARKDOWN_PARSE_ERROR_PATTERN.search(str(error)) is not None + ) + # -- Polling config resolution ------------------------------------------- def resolve_polling_config( diff --git a/src/chat_sdk/thread.py b/src/chat_sdk/thread.py index 1eea4af..5b237cd 100644 --- a/src/chat_sdk/thread.py +++ b/src/chat_sdk/thread.py @@ -9,7 +9,7 @@ import asyncio import contextvars -from collections.abc import AsyncIterator +from collections.abc import AsyncGenerator, AsyncIterator from dataclasses import dataclass, replace from datetime import datetime, timezone from typing import TYPE_CHECKING, Any, Protocol, cast, runtime_checkable @@ -690,7 +690,22 @@ async def _handle_stream( # Build text-only stream from raw_stream text_stream = _from_full_stream(raw_stream) - # Build streaming options from current message context + # Build streaming options from current message context. + # + # Divergence from upstream — see docs/UPSTREAM_SYNC.md. Upstream + # (vercel/chat#340) seeds ``updateIntervalMs`` with the thread-level + # default (``this._streamingUpdateIntervalMs``) before spreading + # caller options, so adapters always see a concrete interval. We + # deliberately leave ``update_interval_ms`` as ``None`` unless the + # caller supplied one: the hand-rolled Teams native streaming path + # treats any non-``None`` value as a caller override of its 1500ms + # quota-protecting emit throttle (upstream's Teams adapter ignores + # the field entirely, so the seed is harmless there). Adapters that + # consume the field apply their own default when it is ``None`` + # (Telegram: ``TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS``), and + # ``_fallback_stream`` already falls back to + # ``self._streaming_update_interval_ms`` — same observable behavior + # as upstream's seed for the fallback path. options = StreamOptions() if self._current_message is not None: options.recipient_user_id = self._current_message.author.user_id @@ -715,7 +730,7 @@ async def _handle_stream( if hasattr(self.adapter, "stream") and self.adapter.stream: # type: ignore[union-attr] accumulated = "" - async def _wrapped_stream() -> AsyncIterator[str | StreamChunk]: + async def _wrapped_stream() -> AsyncGenerator[str | StreamChunk, None]: nonlocal accumulated async for chunk in text_stream: if isinstance(chunk, str): @@ -726,27 +741,38 @@ async def _wrapped_stream() -> AsyncIterator[str | StreamChunk]: accumulated += chunk.text yield chunk - raw_result = await self.adapter.stream(self._id, _wrapped_stream(), options) # type: ignore[union-attr] - # Adapters can override the recorded text via the optional - # ``text`` field on ``RawMessage`` when their internal state - # (cancellation, throttling, partial commits) makes the local - # ``accumulated`` buffer diverge from what the platform - # actually accepted. Default ``None`` falls back to the local - # buffer — backward-compatible for adapters that don't need - # the override (Slack, Discord, GitHub, Google Chat, - # Telegram, Linear, WhatsApp). The Teams native streaming - # path sets it on cancellation to short-circuit the buffered - # suffix that was coalesced into the throttle window but - # never emitted. - recorded_text = raw_result.text if raw_result.text is not None else accumulated - sent = self._create_sent_message( - raw_result.id, - PostableMarkdown(markdown=recorded_text), - raw_result.thread_id, - ) - if self._thread_history is not None: - await self._thread_history.append(self._id, _to_message(sent)) - return sent + wrapped_stream = _wrapped_stream() + raw_result = await self.adapter.stream(self._id, wrapped_stream, options) # type: ignore[union-attr] + if raw_result is not None: + # Adapters can override the recorded text via the optional + # ``text`` field on ``RawMessage`` when their internal state + # (cancellation, throttling, partial commits) makes the local + # ``accumulated`` buffer diverge from what the platform + # actually accepted. Default ``None`` falls back to the local + # buffer — backward-compatible for adapters that don't need + # the override (Slack, Discord, GitHub, Google Chat, + # Telegram, Linear, WhatsApp). The Teams native streaming + # path sets it on cancellation to short-circuit the buffered + # suffix that was coalesced into the throttle window but + # never emitted. + recorded_text = raw_result.text if raw_result.text is not None else accumulated + sent = self._create_sent_message( + raw_result.id, + PostableMarkdown(markdown=recorded_text), + raw_result.thread_id, + ) + if self._thread_history is not None: + await self._thread_history.append(self._id, _to_message(sent)) + return sent + # ``None`` means the adapter delegated back to the SDK's built-in + # post+edit fallback for this thread (vercel/chat#340 — e.g. the + # Telegram adapter only natively streams DMs). The contract + # requires adapters to return ``None`` BEFORE consuming any + # chunks, so ``text_stream`` is still fully intact for the + # fallback below. Close the unused wrapper so a partially + # consumed generator (contract violation) is never left + # suspended awaiting GC finalization. + await wrapped_stream.aclose() # Fallback: post + edit with throttling (text-only) async def _text_only_stream() -> AsyncIterator[str]: diff --git a/src/chat_sdk/types.py b/src/chat_sdk/types.py index 89f7c09..bb84668 100644 --- a/src/chat_sdk/types.py +++ b/src/chat_sdk/types.py @@ -1500,12 +1500,14 @@ async def stream( thread_id: str, text_stream: AsyncIterable[str | StreamChunk], options: StreamOptions | None = None, - ) -> RawMessage: + ) -> RawMessage | None: """Stream a message using platform-native streaming APIs. The adapter consumes the async iterable and handles the entire - streaming lifecycle. Only available on platforms with native - streaming support (e.g., Slack). + streaming lifecycle. Available on platforms with native streaming + or preview APIs. Adapters may return ``None`` before consuming any + chunks to delegate back to the SDK's built-in post+edit fallback + for the current thread (vercel/chat#340). """ raise ChatNotImplementedError(self.name, "stream") diff --git a/tests/test_telegram_streaming.py b/tests/test_telegram_streaming.py new file mode 100644 index 0000000..839309b --- /dev/null +++ b/tests/test_telegram_streaming.py @@ -0,0 +1,502 @@ +"""Tests for Telegram native DM draft streaming and markdown-parse fallback. + +Faithful translation of the ``index.test.ts`` additions from upstream +vercel/chat#340 (commit 5461ea9, "feat(telegram): add native DM draft +streaming with segmented stream results"): ``stream()`` via +``sendMessageDraft`` for private chats, ``None``-delegation for non-DM +threads, and the retry-without-``parse_mode`` path shared by +``post_message`` / ``edit_message`` when Telegram rejects MarkdownV2 +entity parsing. + +Upstream mocks ``fetch`` at the HTTP layer; these tests follow the +established Python convention (see ``test_telegram_api.py``) of mocking +``telegram_fetch`` with an ordered script — each entry is either a value +to return or an exception to raise, mirroring upstream's +``mockResolvedValueOnce`` / ``telegramError`` chains after Bot API error +mapping. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from chat_sdk.adapters.telegram.adapter import ( + TelegramAdapter, +) +from chat_sdk.adapters.telegram.types import TelegramAdapterConfig +from chat_sdk.shared.errors import AdapterRateLimitError, ValidationError +from chat_sdk.shared.mock_adapter import MockLogger + +# ============================================================================= +# Helpers +# ============================================================================= + +DM_THREAD_ID = "telegram:123" +DM_CHAT_ID = "123" + + +def _make_adapter(**overrides: Any) -> TelegramAdapter: + """Create a TelegramAdapter with minimal valid config and a MockLogger.""" + config = TelegramAdapterConfig( + bot_token=overrides.pop("bot_token", "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"), + logger=overrides.pop("logger", MockLogger()), + user_name=overrides.pop("user_name", "mybot"), + **overrides, + ) + return TelegramAdapter(config) + + +def _sample_message( + text: str = "hello world", + message_id: int = 11, + chat_id: int = 123, + chat_type: str = "private", +) -> dict[str, Any]: + """Return a minimal Telegram message dict (mirrors upstream sampleMessage).""" + return { + "message_id": message_id, + "chat": {"id": chat_id, "type": chat_type}, + "from": {"id": 999, "is_bot": True, "first_name": "Bot", "username": "mybot"}, + "date": 1700000000, + "text": text, + } + + +def _parse_entities_error() -> ValidationError: + """The error ``throw_telegram_api_error`` raises for an entity-parse 400.""" + return ValidationError( + "telegram", + "Bad Request: can't parse entities: Can't find end of the entity", + ) + + +def _script_fetch(adapter: TelegramAdapter, script: list[Any]) -> list[tuple[str, Any]]: + """Replace ``telegram_fetch`` with an ordered script of results. + + Each entry is returned in turn; ``Exception`` entries are raised + instead. Returns the recorded ``(method, payload)`` call list. + """ + calls: list[tuple[str, Any]] = [] + queue = list(script) + + async def fetch(method: str, payload: Any = None, **_kwargs: Any) -> Any: + calls.append((method, payload)) + if not queue: + raise AssertionError(f"Unexpected Telegram API call: {method}") + action = queue.pop(0) + if isinstance(action, Exception): + raise action + return action + + adapter.telegram_fetch = fetch # type: ignore[method-assign] + return calls + + +async def _text_stream(chunks: list[str]): + for chunk in chunks: + yield chunk + + +def _stream_options(update_interval_ms: int): + from chat_sdk.types import StreamOptions + + options = StreamOptions() + options.update_interval_ms = update_interval_ms + return options + + +# ============================================================================= +# Tests -- stream() draft streaming (vercel/chat#340) +# ============================================================================= + + +class TestTelegramDraftStreaming: + """it() blocks for TelegramAdapter.stream draft updates.""" + + # it("streams draft updates for private chats and sends a final message") + @pytest.mark.asyncio + async def test_streams_draft_updates_for_private_chats_and_sends_a_final_message(self): + adapter = _make_adapter() + calls = _script_fetch(adapter, [True, True, True, _sample_message()]) + + result = await adapter.stream(DM_THREAD_ID, _text_stream(["hello", " world"]), _stream_options(0)) + + assert result is not None + assert result.id == "123:11" + assert result.thread_id == DM_THREAD_ID + + methods = [method for method, _payload in calls] + assert methods == ["sendMessageDraft", "sendMessageDraft", "sendMessageDraft", "sendMessage"] + + initial_draft = calls[0][1] + first_draft = calls[1][1] + second_draft = calls[2][1] + final_send = calls[3][1] + + assert initial_draft["chat_id"] == DM_CHAT_ID + assert initial_draft["text"] == "" + assert "parse_mode" not in initial_draft + assert first_draft["chat_id"] == DM_CHAT_ID + assert first_draft["draft_id"] == initial_draft["draft_id"] + assert first_draft["text"] == "hello" + assert first_draft["parse_mode"] == "MarkdownV2" + assert second_draft["draft_id"] == first_draft["draft_id"] + assert second_draft["text"] == "hello world" + assert final_send["chat_id"] == DM_CHAT_ID + assert final_send["text"] == "hello world" + assert final_send["parse_mode"] == "MarkdownV2" + + # it("keeps markdown parse mode for an exact-limit draft and final message") + @pytest.mark.asyncio + async def test_keeps_markdown_parse_mode_for_an_exactlimit_draft_and_final_message(self): + long_markdown = "a" * 3494 + "**ok**" + rendered_markdown = "a" * 3494 + "*ok*" + + adapter = _make_adapter() + calls = _script_fetch(adapter, [True, True, _sample_message(text="a" * 3494 + "ok", message_id=41)]) + + result = await adapter.stream( + DM_THREAD_ID, + _text_stream([long_markdown]), + # JS Number.MAX_SAFE_INTEGER: no mid-stream flush; only the + # end-of-stream flush ships the rendered draft. + _stream_options(2**53 - 1), + ) + + assert result is not None + assert [ + (method, len(payload["text"]), payload["text"][-10:], payload.get("parse_mode")) + for method, payload in calls + ] == [ + ("sendMessageDraft", 0, "", None), + ( + "sendMessageDraft", + len(rendered_markdown), + rendered_markdown[-10:], + "MarkdownV2", + ), + ( + "sendMessage", + len(rendered_markdown), + rendered_markdown[-10:], + "MarkdownV2", + ), + ] + + draft_payload = calls[1][1] + final_payload = calls[2][1] + # Exact bodies: no ellipsis was appended and the MarkdownV2-safe + # boundary trim left the balanced ``*ok*`` pair intact. + assert draft_payload["text"] == rendered_markdown + assert final_payload["text"] == rendered_markdown + + # it("returns null for non-DM streaming so Chat SDK can use fallback streaming") + @pytest.mark.asyncio + async def test_returns_null_for_nondm_streaming_so_chat_sdk_can_use_fallback_streaming(self): + adapter = _make_adapter() + calls = _script_fetch(adapter, []) + + result = await adapter.stream( + "telegram:-100123", + _text_stream(["hello"]), + _stream_options(0), + ) + + assert result is None + # Delegation happens before any chunk is consumed or API call made. + assert calls == [] + + # it("falls back to a final message when draft streaming updates fail") + @pytest.mark.asyncio + async def test_falls_back_to_a_final_message_when_draft_streaming_updates_fail(self): + logger = MockLogger() + adapter = _make_adapter(logger=logger) + calls = _script_fetch( + adapter, + [ + ValidationError("telegram", "Bad Request: chat not found"), + _sample_message(), + ], + ) + + result = await adapter.stream(DM_THREAD_ID, _text_stream(["hello", " world"]), _stream_options(0)) + + assert result is not None + assert result.id == "123:11" + assert any( + call[0] == "Telegram draft streaming update failed" and call[1]["thread_id"] == DM_THREAD_ID + for call in logger.warn.calls + ) + + # The failed (non-parse-error) initial draft disables draft + # updates; the stream still persists the final message. + assert [method for method, _payload in calls] == ["sendMessageDraft", "sendMessage"] + + # it("continues to the final message when markdown draft retry also fails") + @pytest.mark.asyncio + async def test_continues_to_the_final_message_when_markdown_draft_retry_also_fails(self): + logger = MockLogger() + adapter = _make_adapter(logger=logger) + calls = _script_fetch( + adapter, + [ + True, # initial empty draft + _parse_entities_error(), # markdown draft rejected + AdapterRateLimitError("telegram", 1), # plain retry 429s + _sample_message(text="**broken"), + ], + ) + + result = await adapter.stream(DM_THREAD_ID, _text_stream(["**broken"]), _stream_options(0)) + + assert result is not None + assert any( + call[0] == "Telegram draft streaming update failed" and call[1]["thread_id"] == DM_THREAD_ID + for call in logger.warn.calls + ) + + final_send = calls[3][1] + assert calls[3][0] == "sendMessage" + assert final_send.get("parse_mode") is None + assert final_send["text"] == "**broken" + + # it("falls back to plain-text draft and final send when Telegram can't + # parse streamed markdown") + @pytest.mark.asyncio + async def test_falls_back_to_plaintext_draft_and_final_send_when_telegram_cant_parse_streamed_markdown(self): + adapter = _make_adapter() + calls = _script_fetch( + adapter, + [ + True, # initial empty draft + _parse_entities_error(), # markdown draft rejected + True, # plain retry accepted + _sample_message(text="**broken"), + ], + ) + + result = await adapter.stream(DM_THREAD_ID, _text_stream(["**broken"]), _stream_options(0)) + + assert result is not None + assert result.id == "123:11" + + retry_draft = calls[2][1] + final_send = calls[3][1] + assert calls[2][0] == "sendMessageDraft" + assert "parse_mode" not in retry_draft + assert retry_draft["text"] == "**broken" + assert calls[3][0] == "sendMessage" + assert final_send.get("parse_mode") is None + assert final_send["text"] == "**broken" + + # it("reuses original text when streamed plain-text fallback would be empty") + @pytest.mark.asyncio + async def test_reuses_original_text_when_streamed_plaintext_fallback_would_be_empty(self): + adapter = _make_adapter() + calls = _script_fetch( + adapter, + [ + True, # initial empty draft + _parse_entities_error(), # markdown draft rejected + True, # plain retry accepted + _sample_message(text="**"), + ], + ) + + result = await adapter.stream(DM_THREAD_ID, _text_stream(["**"]), _stream_options(0)) + + assert result is not None + assert result.id == "123:11" + + # Upstream's remark renders ``**`` to an empty plain string, so the + # fallback resolver reuses the original text. Our parser keeps the + # unpaired markers literal (``**`` plain-renders as ``**``); both + # paths converge on shipping the original text verbatim. The + # empty-fallback branch itself is pinned by + # test_resolve_telegram_fallback_text_reuses_original_when_fallback_is_blank. + retry_draft = calls[2][1] + final_send = calls[3][1] + assert "parse_mode" not in retry_draft + assert retry_draft["text"] == "**" + assert final_send.get("parse_mode") is None + assert final_send["text"] == "**" + + # Python-only: direct adapter.stream call with options omitted entirely. + # Upstream can't hit this via Thread.post (thread.ts seeds + # updateIntervalMs before calling the adapter); our thread.py + # intentionally does NOT seed it (see docs/UPSTREAM_SYNC.md divergence), + # so options=None / update_interval_ms=None must resolve to + # TELEGRAM_DEFAULT_STREAM_UPDATE_INTERVAL_MS instead of crashing. + @pytest.mark.asyncio + async def test_stream_defaults_update_interval_when_options_omitted(self): + adapter = _make_adapter() + calls = _script_fetch(adapter, [True, True, _sample_message()]) + + result = await adapter.stream(DM_THREAD_ID, _text_stream(["hello", " world"])) + + assert result is not None + assert result.id == "123:11" + # Initial empty draft first, persisted message last, and nothing + # but draft updates in between (count depends on wall-clock timing + # against the 250ms default, so only the shape is asserted). + assert calls[0][0] == "sendMessageDraft" + assert calls[0][1]["text"] == "" + assert calls[-1][0] == "sendMessage" + assert all(method == "sendMessageDraft" for method, _payload in calls[:-1]) + + # Python-only: the empty-stream contract from upstream's stream() body + # (`if (!accumulated.trim()) throw`) — whitespace-only streams must + # raise instead of persisting a blank message. + @pytest.mark.asyncio + async def test_stream_rejects_whitespace_only_streams(self): + adapter = _make_adapter() + calls = _script_fetch(adapter, [True]) + + with pytest.raises(ValidationError, match="requires text content"): + await adapter.stream(DM_THREAD_ID, _text_stream([" "]), _stream_options(0)) + + # Only the initial empty draft went out; whitespace renders to "" + # which matches the last draft text, so flushes were skipped. + assert [method for method, _payload in calls] == ["sendMessageDraft"] + + +# ============================================================================= +# Tests -- retry without parse_mode (post_message / edit_message) +# ============================================================================= + + +class TestTelegramMarkdownParseFallback: + """it() blocks for withTelegramMarkdownFallback on post/edit.""" + + # it("retries markdown messages without parse_mode when Telegram can't + # parse entities") + @pytest.mark.asyncio + async def test_retries_markdown_messages_without_parsemode_when_telegram_cant_parse_entities(self): + logger = MockLogger() + adapter = _make_adapter(logger=logger) + calls = _script_fetch( + adapter, + [_parse_entities_error(), _sample_message(text="**broken")], + ) + + result = await adapter.post_message(DM_THREAD_ID, {"markdown": "**broken"}) + + assert result.id == "123:11" + + first_send = calls[0][1] + second_send = calls[1][1] + # First attempt ships the MarkdownV2 rendering (escaped markers). + assert first_send["parse_mode"] == "MarkdownV2" + # Retry drops parse_mode and ships the plain-text rendering. + assert second_send.get("parse_mode") is None + assert second_send["text"] == "**broken" + assert any( + call[0] == "Telegram markdown parse failed; retrying without parse mode" + and call[1]["method"] == "sendMessage" + and call[1]["thread_id"] == DM_THREAD_ID + for call in logger.warn.calls + ) + + # it("retries markdown messages with original text when plain-text + # fallback would be empty") + @pytest.mark.asyncio + async def test_retries_markdown_messages_with_original_text_when_plaintext_fallback_would_be_empty(self): + adapter = _make_adapter() + calls = _script_fetch( + adapter, + [_parse_entities_error(), _sample_message(text="**")], + ) + + result = await adapter.post_message(DM_THREAD_ID, {"markdown": "**"}) + + assert result.id == "123:11" + + second_send = calls[1][1] + assert second_send.get("parse_mode") is None + assert second_send["text"] == "**" + + # it("does not swallow non-parse validation errors during markdown send") + @pytest.mark.asyncio + async def test_does_not_swallow_nonparse_validation_errors_during_markdown_send(self): + adapter = _make_adapter() + calls = _script_fetch( + adapter, + [ValidationError("telegram", "Bad Request: chat not found")], + ) + + with pytest.raises(ValidationError, match="chat not found"): + await adapter.post_message(DM_THREAD_ID, {"markdown": "**broken**"}) + + # No retry attempt for non-parse errors. + assert [method for method, _payload in calls] == ["sendMessage"] + + # it("retries markdown edits without parse_mode when Telegram can't + # parse entities") + @pytest.mark.asyncio + async def test_retries_markdown_edits_without_parsemode_when_telegram_cant_parse_entities(self): + logger = MockLogger() + adapter = _make_adapter(logger=logger) + calls = _script_fetch( + adapter, + [ + _sample_message(text="hello"), # post + _parse_entities_error(), # first edit + _sample_message(text="**broken"), # retry edit + ], + ) + + posted = await adapter.post_message(DM_THREAD_ID, "hello") + result = await adapter.edit_message(DM_THREAD_ID, posted.id, {"markdown": "**broken"}) + + assert result.id == "123:11" + + first_edit = calls[1][1] + second_edit = calls[2][1] + assert calls[1][0] == "editMessageText" + assert first_edit["parse_mode"] == "MarkdownV2" + assert calls[2][0] == "editMessageText" + assert second_edit.get("parse_mode") is None + assert second_edit["text"] == "**broken" + assert any( + call[0] == "Telegram markdown parse failed; retrying without parse mode" + and call[1]["method"] == "editMessageText" + and call[1]["message_id"] == posted.id + and call[1]["thread_id"] == DM_THREAD_ID + for call in logger.warn.calls + ) + + # it("retries markdown edits with original text when plain-text fallback + # would be empty") + @pytest.mark.asyncio + async def test_retries_markdown_edits_with_original_text_when_plaintext_fallback_would_be_empty(self): + adapter = _make_adapter() + calls = _script_fetch( + adapter, + [ + _sample_message(text="hello"), # post + _parse_entities_error(), # first edit + _sample_message(text="**"), # retry edit + ], + ) + + posted = await adapter.post_message(DM_THREAD_ID, "hello") + result = await adapter.edit_message(DM_THREAD_ID, posted.id, {"markdown": "**"}) + + assert result.id == "123:11" + + second_edit = calls[2][1] + assert second_edit.get("parse_mode") is None + assert second_edit["text"] == "**" + + # Python-only: pins the branch upstream exercises through remark's + # empty rendering of ``**`` — our parser keeps those markers literal, + # so the faithful tests above never produce a blank fallback. Without + # this, the ``fallback_text.strip()`` guard could regress unnoticed. + def test_resolve_telegram_fallback_text_reuses_original_when_fallback_is_blank(self): + adapter = _make_adapter() + + assert adapter.resolve_telegram_fallback_text("**original**", "plain") == "plain" + assert adapter.resolve_telegram_fallback_text("**original**", "") == "**original**" + assert adapter.resolve_telegram_fallback_text("**original**", " \n") == "**original**" diff --git a/tests/test_thread_faithful.py b/tests/test_thread_faithful.py index 846a0ff..cffcbd0 100644 --- a/tests/test_thread_faithful.py +++ b/tests/test_thread_faithful.py @@ -387,6 +387,46 @@ async def mock_stream(thread_id, text_stream, options=None): # Local accumulator wins because the adapter didn't override. assert result.text == "Hello world" + # it("should fall back when adapter.stream returns null") + @pytest.mark.asyncio + async def test_should_fall_back_when_adapterstream_returns_null(self): + adapter = create_mock_adapter() + state = create_mock_state() + + mock_stream = AsyncMock(return_value=None) + adapter.stream = mock_stream # type: ignore[attr-defined] + + thread = _make_thread(adapter, state) + text_stream = _create_text_stream(["Hello", " ", "World"]) + await thread.post(text_stream) + + # The adapter was offered the stream first... + mock_stream.assert_called_once() + call_args = mock_stream.call_args + assert call_args.args[0] == "slack:C123:1234.5678" + options = call_args.args[2] + # Divergence from upstream — see docs/UPSTREAM_SYNC.md. Upstream + # asserts ``objectContaining({ updateIntervalMs: 500 })`` here + # because thread.ts (vercel/chat#340) seeds the thread-level + # default into the StreamOptions handed to ``adapter.stream``. We + # intentionally leave ``update_interval_ms`` as ``None`` unless + # caller-supplied: the hand-rolled Teams native streaming path + # treats any non-``None`` value as a caller override of its 1500ms + # quota-protecting emit throttle, so the seed would silently drop + # Teams DM throttling to the 500ms thread default. If this + # assertion fails because seeding was added, reconcile + # ``TeamsAdapter._resolve_emit_interval`` first. + assert options.update_interval_ms is None + + # ...and returning None delegated to the built-in post+edit + # fallback: placeholder posted, then edited with the full text. + assert adapter._post_calls[0] == ("slack:C123:1234.5678", "...") + last_edit = adapter._edit_calls[-1] + assert last_edit[0] == "slack:C123:1234.5678" + assert last_edit[1] == "msg-1" + assert isinstance(last_edit[2], PostableMarkdown) + assert last_edit[2].markdown == "Hello World" + # it("should fall back to post+edit when adapter has no native streaming") @pytest.mark.asyncio async def test_should_fall_back_to_postedit_when_adapter_has_no_native_streaming(self):