diff --git a/PATCHES.md b/PATCHES.md index abae1df0a4c..94a435a1c47 100644 --- a/PATCHES.md +++ b/PATCHES.md @@ -64,7 +64,8 @@ half is upstreamable, the Argus behavior lives in project config). | [#36](#patch-36) | surface `(No response from agent)` on blank final | argus-edit | 89ea4d2f | | [#37](#patch-37) | retry blank final turn + web display guard | argus-edit | 4a19e4fe | | [#38](#patch-38) | per-thread debug-sandbox link | argus-additive | 1aad692a, cd03995e, 2df36c99 (merge) | -| [#39](#patch-39) | checkpointer pool bounds | generic-upstreamable | 348739fc (PR #4, branch `argus-patch-39-pool-bounds`) | +| [#39](#patch-39) | checkpointer pool bounds | generic-upstreamable | f37b8292 (PR #4 squash-merge) | +| [#40](#patch-40) | Telegram send-path extraction to `_telegram_sender.py` | argus-edit (carry-negative refactor) | this PR | Dropped / deferred / not-carried records are at the bottom, followed by the carry budget ledger. @@ -239,7 +240,12 @@ carry budget ledger. sections. "#9" here is the telegram-core re-port label, NOT the dead langgraph_auth patch that older notes also called #9. - Files: `backend/app/channels/_telegram_format.py` (NEW), - `backend/app/channels/telegram.py` (EDITED, heavy 3-way reconciliation), + `backend/app/channels/_telegram_sender.py` (NEW - since patch #40 hosts the + telegram-side send path; state stays on the channel, shims in telegram.py), + `backend/app/channels/telegram.py` (EDITED - since #40 a thin carrier: + send/_send_running_reply/_send_running_reply_safe shims + webhook (#28) + + welcome/lock (#17); upstream's superseded stream helpers restored verbatim + as unreachable code), `backend/app/channels/manager.py` (EDITED), `backend/app/channels/message_bus.py` (EDITED, `OutboundMessage.progress_stage`), `backend/app/gateway/auth_middleware.py` (EDITED, 152a3d5e), @@ -251,11 +257,14 @@ carry budget ledger. - Delete-when: upstream's Telegram channel gains HTML/markdown rendering, a removable working-indicator/stage hook, and per-channel formatter extension points; realistically never as a whole. The exit path is FORK-REVIEW lever - #1: upstream the design or move it behind a cleaner extension point. Both we - and upstream rewrote `telegram.py` with incompatible streaming designs, so - expect a careful 3-way reconciliation on EVERY sync (a naive merge silently - produces broken code). + #1: upstream the design or move it behind a cleaner extension point. Since + patch #40, upstream churn inside its own stream helpers merges clean (they + are restored verbatim, unreachable); the remaining reconciliation surface on + each sync is the three shim bodies, the webhook/welcome edits, and any + upstream change to send()'s signature or OutboundMessage fields (locked by + `test_telegram_sender_seam.py`). - Upstream status: none. +- See also: patch #40 (send-path extraction). ## Patch #10 @@ -720,9 +729,10 @@ carry budget ledger. **Patch #39 - bound the checkpointer pool (elastic 1..4, idle-shrink)** - Class: generic-upstreamable -- Status note: NOT yet on `argus` tip. Lives as fork PR #4, branch - `argus-patch-39-pool-bounds`, commit `348739fc`. Merge it and update this - line in the same commit. +- Status note: merged to `argus` as `f37b8292` (PR #4 squash-merge of branch + `argus-patch-39-pool-bounds`; the pre-merge commit was `348739fc`). This + line was stale until patch #40's PR: the d8ef0d13 ledger rebuild was + authored against 2df36c99, before the merge landed. - Intent: psycopg_pool's default is a fixed pool (max_size falls back to min_size=4), so every uvicorn worker permanently held 4 idle Postgres connections; with 2 workers per gateway and one gateway per project stack, @@ -740,6 +750,51 @@ carry budget ledger. bounds move to config and the code patch drops. - Upstream status: none (fork PR #4; both upstream candidates named above). +## Patch #40 + +**Patch #40 - Telegram send-path extraction to `_telegram_sender.py` (merge-tax reduction)** + +- Class: argus-edit (refactor of #9-chain's telegram.py half; net NEGATIVE + carry - telegram.py drops from 574 to 251 changed lines vs v2.0.0, and + `app/channels/` app-code carry from 1099 to 776) +- Intent: telegram.py was the fork's dominant merge tax (FORK-REVIEW lever + #1). Zero behavior change (the pre-existing suite, unmodified, is the + proof): the argus send path - stage-emoji indicator (show/promote/clear), + HTML chunked sends with retry + plain-text fallback - moves verbatim into + argus-owned `_telegram_sender.py` as channel-first free functions; ALL + mutable state stays on the channel instance (`init_state(channel, config)` + called from `__init__`) because tests read/patch `ch._working_msg` & friends + directly, and telegram.py keeps three bound-method shims (`send`, + `_send_running_reply`, `_send_running_reply_safe`) because tests and the + receive path override/dispatch per instance (`_send_running_reply_safe` goes + through `self._send_running_reply`, never the module function). Upstream + v2.0.0's superseded edit-in-place stream helpers (`_send_stream_update` .. + `_split_message`, their module constants, `_monotonic`, and the + `_stream_messages` init) are RESTORED byte-identical but unreachable - + `send()` never routes to them - so upstream churn in those regions merges + clean instead of modify/delete-conflicting on every sync. Numbering note: + inline `[argus patch #10]` comments inside the moved bodies are historical + labels for the #9-chain stage-emoji work; left verbatim on purpose. +- Files: `backend/app/channels/_telegram_sender.py` (NEW), + `backend/app/channels/telegram.py` (EDITED - shrunk; residual argus content + is the #17 welcome/lock, the #28 webhook mode, the three #40 shims + + `init_state` call, and the `_log_future_error` static override noted below) +- Tests: `backend/tests/test_telegram_sender_seam.py` (NEW - delegation + + contract locks so a naive future merge that resurrects upstream's send body + fails loudly); behavior proof is the UNTOUCHED existing suite + (`test_telegram_send.py`, `test_channels.py` Telegram classes, + `test_telegram_channel_connections.py`). +- Delete-when: with #9-chain (this module hosts its telegram-side behavior). + If #9-chain is upstreamed or re-expressed behind an upstream extension + point, `_telegram_sender.py` goes with it and telegram.py returns to + vanilla + #17/#28. +- Upstream status: n/a (fork-internal restructuring). +- Follow-up candidate (NOT in #40, would change log wording): telegram.py's + static `_log_future_error` shadows the base-class method upstream v2.0.0 + already provides; deleting the override would shave ~9 more carried lines + but is a real (if tiny) behavior delta - out of scope for a zero-behavior + patch. + --- ## Dropped / deferred / re-expressed (v2.0.0 rebase record - do not re-add blindly) @@ -802,3 +857,13 @@ design or move it behind an extension point). | Date | Base | Commits | Files | Lines | Upstream-file edited lines | |---|---|---|---|---|---| | 2026-07-01 | v2.0.0 -> 2df36c99 | 29 | 85 | +6168 / -812 | ~1600 (~1460 in `app/channels/`) | +| 2026-07-02 | v2.0.0 -> #40 tip | 32 | 90 | +7433 / -668 | app-code excl. tests/docs: 1923 (776 in `app/channels/`, was 1099); tests: 1350. #40 cut `telegram.py` 574 -> 251 | + +Methodology note (2026-07-02): the last column is now measured against the +`v2.0.0` tag over files that exist at v2.0.0 (insertions+deletions), split +app-code vs tests. The 2026-07-01 row's ~1600/~1460 came from FORK-REVIEW's +2026-06-30 measurement against merge-base 2ace78d1 with a different file +scope and is not directly comparable; like-for-like against v2.0.0 the +pre-#40 tip was 2246 app-code (1099 in `app/channels/`). Reproduce with: +`git diff --numstat v2.0.0 | while read a d f; do git cat-file -e +"v2.0.0:$f" 2>/dev/null && echo "$a $d $f"; done | awk '...'`. diff --git a/backend/app/channels/_telegram_sender.py b/backend/app/channels/_telegram_sender.py new file mode 100644 index 00000000000..eccd5c7b1b3 --- /dev/null +++ b/backend/app/channels/_telegram_sender.py @@ -0,0 +1,294 @@ +"""[argus patch #40] Telegram send path: stage-emoji indicator + HTML chunked sends. + +Extracted verbatim from app/channels/telegram.py to shrink the fork's +upstream-edited-line carry (the dominant merge tax). telegram.py keeps thin +bound-method shims (send / _send_running_reply / _send_running_reply_safe) +plus upstream's own code; everything argus-specific about SENDING lives here. + +Behavior (ledgered as patch #9-chain): + * a progress_stage OutboundMessage renders as an animated lone-emoji message, + deleted + re-sent on stage change (Telegram animates only on first send), + throttled to stage_min_interval, with a received->thinking auto-promote; + * chat text converts markdown -> Telegram HTML (_telegram_format) and chunks + at 4096, with retry + a plain-text fallback when Telegram rejects the HTML; + * only the final message clears the indicator. + +Free functions take the TelegramChannel as first argument; ALL mutable state +lives on the channel instance (see init_state) because tests read/patch +ch._working_msg & friends directly, and per-instance overrides of +ch._send_running_reply must keep working (send_running_reply_safe dispatches +through the bound method, never the module function). The bot is resolved from +channel._application at call time — it is None until the channel's start(). + +Historical note: inline "[argus patch #10]" markers in the moved bodies are +the pre-ledger label for the #9-chain stage-emoji work and are left verbatim +on purpose; this extraction is patch #40 — see PATCHES.md. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from typing import TYPE_CHECKING, Any + +from app.channels._telegram_format import chunk_html, to_telegram_html + +if TYPE_CHECKING: + from app.channels.message_bus import OutboundMessage + from app.channels.telegram import TelegramChannel + +logger = logging.getLogger(__name__) + +# [argus patch #10] Live stage-emoji progress indicator. Telegram renders a +# lone-emoji message large + animated, but the animation plays ONCE on send +# and never replays on edit (core.telegram.org/api/animated-emojis). So to +# animate each stage we DELETE the old emoji message and SEND a new one when +# the agent's stage changes. The manager derives stages from the langgraph +# stream; this maps each to an emoji. Re-sends are throttled to >= the +# animation length so each one completes. Overridable via +# channels.telegram.{stage_emoji (map), stage_min_interval, working_emoji}. +_DEFAULT_STAGE_EMOJI = { + "received": "👀", + "thinking": "🧠", + "planning": "📝", + "searching": "🔍", + "working": "🔧", + "writing": "✍️", +} +_DEFAULT_STAGE_MIN_INTERVAL = 6.0 # seconds; matches the ~6s big-emoji animation + + +def init_state(channel: TelegramChannel, config: dict[str, Any]) -> None: + """Initialize the sender's per-chat state ON the channel instance and + parse its config knobs. Called from TelegramChannel.__init__; must not + touch channel._application (set later, in start()).""" + # [argus patch #10] Live stage-emoji indicator state, per chat: + # _working_msg[chat] -> message_id of the current stage emoji + # _working_stage[chat] -> the stage that emoji represents + # _working_at[chat] -> monotonic time we last (re)sent it + channel._working_msg = {} + channel._working_stage = {} + channel._working_at = {} + # chat -> task that auto-promotes 👀 received → 🧠 thinking after + # stage_min_interval if no real stage signal has arrived. Guarantees the + # indicator always advances even when the agent reports no tool stage. + channel._promote_timer = {} + # Stage → emoji map (config override merges over the defaults). + channel._stage_emoji = dict(_DEFAULT_STAGE_EMOJI) + cfg_map = config.get("stage_emoji") + if isinstance(cfg_map, dict): + channel._stage_emoji.update({str(k): str(v) for k, v in cfg_map.items()}) + # Back-compat: a bare working_emoji still overrides the initial beat. + if config.get("working_emoji"): + channel._stage_emoji["received"] = str(config["working_emoji"]) + try: + channel._stage_min_interval = float(config.get("stage_min_interval", _DEFAULT_STAGE_MIN_INTERVAL)) + except (ValueError, TypeError): + channel._stage_min_interval = _DEFAULT_STAGE_MIN_INTERVAL + + +async def send(channel: TelegramChannel, msg: OutboundMessage, *, max_retries: int = 3) -> None: + if not channel._application: + return + + try: + chat_id = int(msg.chat_id) + except (ValueError, TypeError): + logger.error("Invalid Telegram chat_id: %s", msg.chat_id) + return + + # [argus patch #10] A progress signal is not chat content — render it as + # the animated stage emoji and return BEFORE the HTML/chunk send path. + if msg.progress_stage is not None: + await show_stage(channel, chat_id, msg.chat_id, msg.progress_stage) + return + + # [argus patch #10] Convert the agent's markdown to Telegram-native HTML + # and split on the 4096-char ceiling without breaking tags. Empty text + # (e.g. an attachment-only message) sends nothing here; send_file + # handles the upload separately. + html = to_telegram_html(msg.text) if msg.text else "" + chunks = chunk_html(html) if html else [] + + bot = channel._application.bot + for chunk in chunks: + await send_one(channel, bot, chat_id, msg.chat_id, chunk, max_retries=max_retries) + + # [argus patch #10] Only the final message in a response clears the + # working-indicator emoji. Streaming partials (is_final=False) leave it. + if msg.is_final: + await clear_working(channel, chat_id, msg.chat_id) + + +async def send_one(channel: TelegramChannel, bot, chat_id: int, chat_key: str, text: str, *, max_retries: int = 3) -> None: + """Send a single (already chunked) HTML message with retry + a + plain-text fallback if Telegram rejects the HTML (a malformed entity + must never drop the message).""" + kwargs: dict[str, Any] = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"} + reply_to = channel._last_bot_message.get(chat_key) + if reply_to: + kwargs["reply_to_message_id"] = reply_to + + last_exc: Exception | None = None + for attempt in range(max_retries): + try: + sent = await bot.send_message(**kwargs) + channel._last_bot_message[chat_key] = sent.message_id + return + except Exception as exc: + last_exc = exc + # A BadRequest is almost always malformed HTML — retrying the + # same HTML won't help, so drop parse_mode and resend as plain + # text on the next attempt (mirrors the ateam fallback). + if exc.__class__.__name__ == "BadRequest" and kwargs.get("parse_mode"): + logger.warning("[Telegram] HTML rejected (%s); resending as plain text", exc) + kwargs.pop("parse_mode", None) + continue + if attempt < max_retries - 1: + delay = 2**attempt # 1s, 2s + logger.warning( + "[Telegram] send failed (attempt %d/%d), retrying in %ds: %s", + attempt + 1, + max_retries, + delay, + exc, + ) + await asyncio.sleep(delay) + + logger.error("[Telegram] send failed after %d attempts: %s", max_retries, last_exc) + if last_exc is None: + # Degenerate config (_max_retries=0): the loop never ran, so there's + # no exception to propagate. Surface it explicitly rather than + # silently succeeding. (Preserves the pre-patch contract.) + raise RuntimeError("Telegram send failed without an exception from any attempt") + raise last_exc + + +async def send_running_reply(channel: TelegramChannel, chat_id: str, reply_to_message_id: int) -> None: + """[argus patch #10] Show the initial 'received' stage emoji as soon as + a message arrives. Subsequent stages come from the manager as + progress_stage OutboundMessages and are handled by show_stage. If the + emoji send fails, fall back to a reaction on the user's message.""" + if not channel._application: + return + ok = await show_stage(channel, int(chat_id), chat_id, "received") + if ok: + return + # Degraded mode: react on the user's message instead. + try: + from telegram import ReactionTypeEmoji + + await channel._application.bot.set_message_reaction( + chat_id=int(chat_id), + message_id=reply_to_message_id, + reaction=[ReactionTypeEmoji(channel._stage_emoji.get("received", "👀"))], + ) + except Exception: + logger.warning("[Telegram] reaction fallback also failed in chat=%s", chat_id) + + +async def send_running_reply_safe(channel: TelegramChannel, chat_id: str, msg_id: int) -> None: + """Fire-and-forget wrapper for _send_running_reply that logs errors. + + Dispatches through the BOUND channel._send_running_reply (never the + module-level send_running_reply) so per-instance overrides of that + method — a documented test seam — keep intercepting.""" + try: + await channel._send_running_reply(chat_id, msg_id) + except Exception: + logger.exception("[Telegram] fire-and-forget _send_running_reply failed in chat=%s", chat_id) + + +async def show_stage(channel: TelegramChannel, chat_id: int, chat_key: str, stage: str, *, force: bool = False) -> bool: + """[argus patch #10] Render a stage as the animated lone emoji. + + The big-emoji animation only plays on first SEND (an edit never + replays it), so to animate each new stage we delete the prior emoji + message and send a fresh one. Re-send only when (a) the stage's emoji + actually changes and (b) at least stage_min_interval has elapsed since + the last send — so each animation completes and we stay well under + Telegram's ~1 msg/sec per-chat limit. Rapid intermediate stages are + skipped (latest-wins). ``force`` bypasses the interval guard (used by + the 👀→🧠 auto-promote timer, which already waited the interval). + Returns True if an emoji is now showing. + """ + if not channel._application: + return False + emoji = channel._stage_emoji.get(stage) + if not emoji: + return bool(channel._working_msg.get(chat_key)) + + now = time.monotonic() + cur_stage = channel._working_stage.get(chat_key) + have_msg = chat_key in channel._working_msg + + # No change, or too soon since the last (re)send → leave it be. The next + # eligible stage signal will carry the then-current stage (latest-wins). + if have_msg and cur_stage == stage: + return True + if have_msg and not force and (now - channel._working_at.get(chat_key, 0.0)) < channel._stage_min_interval: + return True + + bot = channel._application.bot + old_id = channel._working_msg.get(chat_key) + try: + sent = await bot.send_message(chat_id=chat_id, text=emoji) + except Exception: + logger.exception("[Telegram] failed to send stage emoji in chat=%s", chat_key) + return have_msg + channel._working_msg[chat_key] = sent.message_id + channel._working_stage[chat_key] = stage + channel._working_at[chat_key] = now + logger.info("[Telegram] stage %s (%s) in chat=%s", stage, emoji, chat_key) + # Delete the previous emoji after the new one is up (no visible gap). + if old_id is not None: + try: + await bot.delete_message(chat_id=chat_id, message_id=old_id) + except Exception as exc: + logger.debug("[Telegram] could not delete prior stage emoji in chat=%s: %s", chat_key, exc) + # Any real stage supersedes a pending auto-promote; 'received' (re)arms it. + cancel_promote(channel, chat_key) + if stage == "received": + schedule_promote(channel, chat_id, chat_key) + return True + + +def schedule_promote(channel: TelegramChannel, chat_id: int, chat_key: str) -> None: + """Arm the 👀→🧠 auto-promotion so the indicator always advances even + if the agent never reports a tool stage.""" + channel._promote_timer[chat_key] = asyncio.ensure_future(auto_promote(channel, chat_id, chat_key)) + + +def cancel_promote(channel: TelegramChannel, chat_key: str) -> None: + task = channel._promote_timer.pop(chat_key, None) + if task and not task.done(): + task.cancel() + + +async def auto_promote(channel: TelegramChannel, chat_id: int, chat_key: str) -> None: + try: + await asyncio.sleep(channel._stage_min_interval) + except asyncio.CancelledError: + return + # Drop our own handle first so show_stage's cancel_promote (which runs + # for every send) doesn't cancel this still-running task mid-flight. + channel._promote_timer.pop(chat_key, None) + # Only promote if we're still on 'received' (no real stage arrived). + if channel._working_stage.get(chat_key) == "received": + await show_stage(channel, chat_id, chat_key, "thinking", force=True) + + +async def clear_working(channel: TelegramChannel, chat_id: int, chat_key: str) -> None: + """[argus patch #10] Delete the current stage emoji once the final + answer has been sent. Best-effort — the message may already be gone.""" + cancel_promote(channel, chat_key) + channel._working_stage.pop(chat_key, None) + channel._working_at.pop(chat_key, None) + msg_id = channel._working_msg.pop(chat_key, None) + if msg_id is None: + return + try: + await channel._application.bot.delete_message(chat_id=chat_id, message_id=msg_id) + except Exception as exc: + logger.debug("[Telegram] could not delete stage emoji in chat=%s: %s", chat_key, exc) diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index 5cbf2727f6a..1e252729ea9 100644 --- a/backend/app/channels/telegram.py +++ b/backend/app/channels/telegram.py @@ -20,30 +20,32 @@ from fastapi import Request, Response -from app.channels._telegram_format import chunk_html, to_telegram_html +from app.channels import _telegram_sender from app.channels.base import Channel from app.channels.connection_identity import attach_connection_identity from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment logger = logging.getLogger(__name__) -# [argus patch #10] Live stage-emoji progress indicator. Telegram renders a -# lone-emoji message large + animated, but the animation plays ONCE on send -# and never replays on edit (core.telegram.org/api/animated-emojis). So to -# animate each stage we DELETE the old emoji message and SEND a new one when -# the agent's stage changes. The manager derives stages from the langgraph -# stream; this maps each to an emoji. Re-sends are throttled to >= the -# animation length so each one completes. Overridable via -# channels.telegram.{stage_emoji (map), stage_min_interval, working_emoji}. -_DEFAULT_STAGE_EMOJI = { - "received": "👀", - "thinking": "🧠", - "planning": "📝", - "searching": "🔍", - "working": "🔧", - "writing": "✍️", -} -_DEFAULT_STAGE_MIN_INTERVAL = 6.0 # seconds; matches the ~6s big-emoji animation +TELEGRAM_MAX_MESSAGE_LENGTH = 4096 +STREAM_EDIT_MIN_INTERVAL_SECONDS = 1.0 +# Groups (negative chat_id) are capped at 20 messages/minute by Telegram, +# so stream edits there must pace well below the private-chat 1 msg/s guideline. +STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS = 3.0 +# Bound on tracked in-flight streamed messages; entries normally clear on the +# final update, this only guards against leaks when a final never arrives. +MAX_TRACKED_STREAM_MESSAGES = 256 + +# Indirection so tests can patch the clock without touching the global time module. +_monotonic = time.monotonic + +# [argus patch #40] The constants above and the class's stream-edit helpers +# (_send_stream_update through _split_message, and the _stream_messages init) +# are upstream v2.0.0's edit-in-place streaming path, restored byte-identical +# but UNREACHABLE: send() delegates to app.channels._telegram_sender (the +# stage-emoji + HTML design, patch #9-chain). Kept verbatim so upstream churn +# in these regions merges clean instead of modify/delete-conflicting on every +# sync. Do not call them; do not "clean them up". See PATCHES.md #40. class TelegramChannel(Channel): @@ -58,17 +60,6 @@ class TelegramChannel(Channel): ``X-Telegram-Bot-Api-Secret-Token`` header doesn't match. """ - # [argus patch #10] Take the manager's streaming path so we receive - # progress_stage signals to drive the animated stage emoji. The manager's - # _channel_supports_streaming reads THIS property (not CHANNEL_CAPABILITIES) - # whenever a live channel object exists, so the capability flip alone is not - # enough — this override is what actually engages streaming. (Telegram still - # gets only stage signals + the final answer, never streamed partial text; - # that suppression lives in the manager.) - @property - def supports_streaming(self) -> bool: - return True - def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None: super().__init__(name="telegram", bus=bus, config=config) self._application = None @@ -83,34 +74,28 @@ def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None: pass # chat_id -> last sent message_id for threaded replies self._last_bot_message: dict[str, int] = {} - # [argus patch #10] Live stage-emoji indicator state, per chat: - # _working_msg[chat] -> message_id of the current stage emoji - # _working_stage[chat] -> the stage that emoji represents - # _working_at[chat] -> monotonic time we last (re)sent it - self._working_msg: dict[str, int] = {} - self._working_stage: dict[str, str] = {} - self._working_at: dict[str, float] = {} - # chat -> task that auto-promotes 👀 received → 🧠 thinking after - # stage_min_interval if no real stage signal has arrived. Guarantees the - # indicator always advances even when the agent reports no tool stage. - self._promote_timer: dict[str, asyncio.Task] = {} - # Stage → emoji map (config override merges over the defaults). - self._stage_emoji: dict[str, str] = dict(_DEFAULT_STAGE_EMOJI) - cfg_map = config.get("stage_emoji") - if isinstance(cfg_map, dict): - self._stage_emoji.update({str(k): str(v) for k, v in cfg_map.items()}) - # Back-compat: a bare working_emoji still overrides the initial beat. - if config.get("working_emoji"): - self._stage_emoji["received"] = str(config["working_emoji"]) - try: - self._stage_min_interval: float = float(config.get("stage_min_interval", _DEFAULT_STAGE_MIN_INTERVAL)) - except (ValueError, TypeError): - self._stage_min_interval = _DEFAULT_STAGE_MIN_INTERVAL + # stream_key ("chat_id:thread_ts") -> state of the in-flight streamed + # bot message being edited in place: {"message_id", "last_edit_at", "last_text"} + self._stream_messages: dict[str, dict[str, Any]] = {} + # [argus patch #40] Stage-emoji indicator state + config parsing + # (stage_emoji / working_emoji / stage_min_interval) — see _telegram_sender. + _telegram_sender.init_state(self, config) # Webhook mode config. self._webhook_mode: bool = bool(config.get("webhook", False)) self._webhook_secret: str = str(config.get("webhook_secret", "")) + # [argus patch #10] Take the manager's streaming path so we receive + # progress_stage signals to drive the animated stage emoji. The manager's + # _channel_supports_streaming reads THIS property (not CHANNEL_CAPABILITIES) + # whenever a live channel object exists, so the capability flip alone is not + # enough — this override is what actually engages streaming. (Telegram still + # gets only stage signals + the final answer, never streamed partial text; + # that suppression lives in the manager.) + @property + def supports_streaming(self) -> bool: + return True + async def start(self) -> None: if self._running: return @@ -186,79 +171,126 @@ async def stop(self) -> None: logger.info("Telegram channel stopped") async def send(self, msg: OutboundMessage, *, _max_retries: int = 3) -> None: - if not self._application: + # [argus patch #40] The argus send path (stage emoji + HTML chunking, + # patch #9-chain) lives in app.channels._telegram_sender; this shim is + # the Channel ABC entry point. Signature — including the test-facing + # _max_retries kwarg — matches upstream v2.0.0. The stream-edit helpers + # below are upstream's superseded path, kept verbatim but unreachable. + await _telegram_sender.send(self, msg, max_retries=_max_retries) + + async def _send_stream_update(self, chat_id: int, key: str, text: str, reply_to: int | None = None) -> None: + """Edit the in-flight streamed message with accumulated text. + + Updates are best-effort: throttled, rate-limit drops are silent. The + manager always publishes a final message afterwards, which guarantees + delivery of the complete text. + """ + if not text: return - try: - chat_id = int(msg.chat_id) - except (ValueError, TypeError): - logger.error("Invalid Telegram chat_id: %s", msg.chat_id) + display = text + if len(display) > TELEGRAM_MAX_MESSAGE_LENGTH: + display = display[: TELEGRAM_MAX_MESSAGE_LENGTH - 1] + "…" + + bot = self._application.bot + state = self._stream_messages.get(key) + + send_kwargs: dict[str, Any] = {"chat_id": chat_id, "text": display} + if reply_to: + send_kwargs["reply_to_message_id"] = reply_to + + if state is None: + try: + sent = await bot.send_message(**send_kwargs) + except Exception: + logger.exception("[Telegram] failed to start stream message in chat=%s", chat_id) + return + self._register_stream_message(key, message_id=sent.message_id, last_text=display, last_edit_at=_monotonic()) return - # [argus patch #10] A progress signal is not chat content — render it as - # the animated stage emoji and return BEFORE the HTML/chunk send path. - if msg.progress_stage is not None: - await self._show_stage(chat_id, msg.chat_id, msg.progress_stage) + now = _monotonic() + min_interval = STREAM_EDIT_GROUP_MIN_INTERVAL_SECONDS if chat_id < 0 else STREAM_EDIT_MIN_INTERVAL_SECONDS + if now - state["last_edit_at"] < min_interval: + return + if display == state["last_text"]: return - # [argus patch #10] Convert the agent's markdown to Telegram-native HTML - # and split on the 4096-char ceiling without breaking tags. Empty text - # (e.g. an attachment-only message) sends nothing here; send_file - # handles the upload separately. - html = to_telegram_html(msg.text) if msg.text else "" - chunks = chunk_html(html) if html else [] + try: + await bot.edit_message_text(chat_id=chat_id, message_id=state["message_id"], text=display) + except Exception as exc: + if self._is_not_modified(exc): + state["last_text"] = display + return + if self._is_retry_after(exc): + logger.debug("[Telegram] stream edit rate-limited in chat=%s, dropping update", chat_id) + return + logger.warning("[Telegram] stream edit failed in chat=%s, sending new message: %s", chat_id, exc) + try: + sent = await bot.send_message(**send_kwargs) + except Exception: + logger.exception("[Telegram] failed to send fallback stream message in chat=%s", chat_id) + return + state["message_id"] = sent.message_id + + state["last_edit_at"] = _monotonic() + state["last_text"] = display + async def _finalize_stream_message(self, chat_id: int, chat_key: str, state: dict[str, Any], text: str) -> None: + """Apply the final text: edit the streamed message, splitting overflow into follow-ups.""" bot = self._application.bot - for chunk in chunks: - await self._send_one(bot, chat_id, msg.chat_id, chunk, _max_retries=_max_retries) - - # [argus patch #10] Only the final message in a response clears the - # working-indicator emoji. Streaming partials (is_final=False) leave it. - if msg.is_final: - await self._clear_working(chat_id, msg.chat_id) - - async def _send_one(self, bot, chat_id: int, chat_key: str, text: str, *, _max_retries: int = 3) -> None: - """Send a single (already chunked) HTML message with retry + a - plain-text fallback if Telegram rejects the HTML (a malformed entity - must never drop the message).""" - kwargs: dict[str, Any] = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"} - reply_to = self._last_bot_message.get(chat_key) - if reply_to: - kwargs["reply_to_message_id"] = reply_to + chunks = self._split_message(text or "") - last_exc: Exception | None = None - for attempt in range(_max_retries): + edited = True + if chunks[0] != state["last_text"]: + edited = await self._edit_final_chunk(bot, chat_id, state["message_id"], chunks[0]) + + if edited: + self._last_bot_message[chat_key] = state["message_id"] + else: + # Edit could not be applied (e.g. message deleted) — deliver the + # first chunk as a fresh message with the standard retry policy. + await self._send_new_message(chat_id, chat_key, chunks[0]) + + for chunk in chunks[1:]: + await self._send_new_message(chat_id, chat_key, chunk) + + async def _edit_final_chunk(self, bot, chat_id: int, message_id: int, text: str) -> bool: + """Edit with one rate-limit retry. Returns False if the edit could not be applied.""" + for attempt in range(2): try: - sent = await bot.send_message(**kwargs) - self._last_bot_message[chat_key] = sent.message_id - return + await bot.edit_message_text(chat_id=chat_id, message_id=message_id, text=text) + return True except Exception as exc: - last_exc = exc - # A BadRequest is almost always malformed HTML — retrying the - # same HTML won't help, so drop parse_mode and resend as plain - # text on the next attempt (mirrors the ateam fallback). - if exc.__class__.__name__ == "BadRequest" and kwargs.get("parse_mode"): - logger.warning("[Telegram] HTML rejected (%s); resending as plain text", exc) - kwargs.pop("parse_mode", None) + if self._is_not_modified(exc): + return True + if self._is_retry_after(exc) and attempt == 0: + await asyncio.sleep(self._retry_after_seconds(exc)) continue - if attempt < _max_retries - 1: - delay = 2**attempt # 1s, 2s - logger.warning( - "[Telegram] send failed (attempt %d/%d), retrying in %ds: %s", - attempt + 1, - _max_retries, - delay, - exc, - ) - await asyncio.sleep(delay) - - logger.error("[Telegram] send failed after %d attempts: %s", _max_retries, last_exc) - if last_exc is None: - # Degenerate config (_max_retries=0): the loop never ran, so there's - # no exception to propagate. Surface it explicitly rather than - # silently succeeding. (Preserves the pre-patch contract.) - raise RuntimeError("Telegram send failed without an exception from any attempt") - raise last_exc + logger.warning("[Telegram] final edit failed in chat=%s: %s", chat_id, exc) + return False + return False + + async def _send_new_message(self, chat_id: int, chat_key: str, text: str, *, _max_retries: int = 3) -> int | None: + """Send a fresh message with retry/backoff. Returns the sent message_id.""" + kwargs: dict[str, Any] = {"chat_id": chat_id, "text": text} + + # Reply to the last bot message in this chat for threading + reply_to = self._last_bot_message.get(chat_key) + if reply_to: + kwargs["reply_to_message_id"] = reply_to + + bot = self._application.bot + + async def send_message() -> int: + sent = await bot.send_message(**kwargs) + self._last_bot_message[chat_key] = sent.message_id + return sent.message_id + + return await self._send_with_retry( + send_message, + max_retries=_max_retries, + log_prefix="[Telegram]", + ) async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) -> bool: if not self._application: @@ -304,116 +336,52 @@ async def send_file(self, msg: OutboundMessage, attachment: ResolvedAttachment) # -- helpers ----------------------------------------------------------- - async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) -> None: - """[argus patch #10] Show the initial 'received' stage emoji as soon as - a message arrives. Subsequent stages come from the manager as - progress_stage OutboundMessages and are handled by _show_stage. If the - emoji send fails, fall back to a reaction on the user's message.""" - if not self._application: - return - ok = await self._show_stage(int(chat_id), chat_id, "received") - if ok: - return - # Degraded mode: react on the user's message instead. - try: - from telegram import ReactionTypeEmoji - - await self._application.bot.set_message_reaction( - chat_id=int(chat_id), - message_id=reply_to_message_id, - reaction=[ReactionTypeEmoji(self._stage_emoji.get("received", "👀"))], - ) - except Exception: - logger.warning("[Telegram] reaction fallback also failed in chat=%s", chat_id) - - async def _show_stage(self, chat_id: int, chat_key: str, stage: str, *, force: bool = False) -> bool: - """[argus patch #10] Render a stage as the animated lone emoji. - - The big-emoji animation only plays on first SEND (an edit never - replays it), so to animate each new stage we delete the prior emoji - message and send a fresh one. Re-send only when (a) the stage's emoji - actually changes and (b) at least stage_min_interval has elapsed since - the last send — so each animation completes and we stay well under - Telegram's ~1 msg/sec per-chat limit. Rapid intermediate stages are - skipped (latest-wins). ``force`` bypasses the interval guard (used by - the 👀→🧠 auto-promote timer, which already waited the interval). - Returns True if an emoji is now showing. - """ - if not self._application: - return False - emoji = self._stage_emoji.get(stage) - if not emoji: - return bool(self._working_msg.get(chat_key)) + @staticmethod + def _stream_key(chat_id: str, thread_ts: str | None) -> str: + return f"{chat_id}:{thread_ts or ''}" - now = time.monotonic() - cur_stage = self._working_stage.get(chat_key) - have_msg = chat_key in self._working_msg + @staticmethod + def _parse_message_id(value: str | None) -> int | None: + try: + return int(value) if value else None + except (TypeError, ValueError): + return None + + def _register_stream_message(self, key: str, *, message_id: int, last_text: str, last_edit_at: float) -> None: + self._stream_messages.pop(key, None) + while len(self._stream_messages) >= MAX_TRACKED_STREAM_MESSAGES: + self._stream_messages.pop(next(iter(self._stream_messages))) + self._stream_messages[key] = { + "message_id": message_id, + "last_edit_at": last_edit_at, + "last_text": last_text, + } - # No change, or too soon since the last (re)send → leave it be. The next - # eligible stage signal will carry the then-current stage (latest-wins). - if have_msg and cur_stage == stage: - return True - if have_msg and not force and (now - self._working_at.get(chat_key, 0.0)) < self._stage_min_interval: - return True + @staticmethod + def _is_retry_after(exc: Exception) -> bool: + return getattr(exc, "retry_after", None) is not None - bot = self._application.bot - old_id = self._working_msg.get(chat_key) - try: - sent = await bot.send_message(chat_id=chat_id, text=emoji) - except Exception: - logger.exception("[Telegram] failed to send stage emoji in chat=%s", chat_key) - return have_msg - self._working_msg[chat_key] = sent.message_id - self._working_stage[chat_key] = stage - self._working_at[chat_key] = now - logger.info("[Telegram] stage %s (%s) in chat=%s", stage, emoji, chat_key) - # Delete the previous emoji after the new one is up (no visible gap). - if old_id is not None: - try: - await bot.delete_message(chat_id=chat_id, message_id=old_id) - except Exception as exc: - logger.debug("[Telegram] could not delete prior stage emoji in chat=%s: %s", chat_key, exc) - # Any real stage supersedes a pending auto-promote; 'received' (re)arms it. - self._cancel_promote(chat_key) - if stage == "received": - self._schedule_promote(chat_id, chat_key) - return True + @staticmethod + def _retry_after_seconds(exc: Exception) -> float: + value = getattr(exc, "retry_after", 0) + if hasattr(value, "total_seconds"): + return float(value.total_seconds()) + return float(value) - def _schedule_promote(self, chat_id: int, chat_key: str) -> None: - """Arm the 👀→🧠 auto-promotion so the indicator always advances even - if the agent never reports a tool stage.""" - self._promote_timer[chat_key] = asyncio.ensure_future(self._auto_promote(chat_id, chat_key)) + @staticmethod + def _is_not_modified(exc: Exception) -> bool: + return "message is not modified" in str(exc).lower() - def _cancel_promote(self, chat_key: str) -> None: - task = self._promote_timer.pop(chat_key, None) - if task and not task.done(): - task.cancel() + @staticmethod + def _split_message(text: str) -> list[str]: + return [text[i : i + TELEGRAM_MAX_MESSAGE_LENGTH] for i in range(0, len(text), TELEGRAM_MAX_MESSAGE_LENGTH)] or [text] - async def _auto_promote(self, chat_id: int, chat_key: str) -> None: - try: - await asyncio.sleep(self._stage_min_interval) - except asyncio.CancelledError: - return - # Drop our own handle first so _show_stage's _cancel_promote (which runs - # for every send) doesn't cancel this still-running task mid-flight. - self._promote_timer.pop(chat_key, None) - # Only promote if we're still on 'received' (no real stage arrived). - if self._working_stage.get(chat_key) == "received": - await self._show_stage(chat_id, chat_key, "thinking", force=True) - - async def _clear_working(self, chat_id: int, chat_key: str) -> None: - """[argus patch #10] Delete the current stage emoji once the final - answer has been sent. Best-effort — the message may already be gone.""" - self._cancel_promote(chat_key) - self._working_stage.pop(chat_key, None) - self._working_at.pop(chat_key, None) - msg_id = self._working_msg.pop(chat_key, None) - if msg_id is None: - return - try: - await self._application.bot.delete_message(chat_id=chat_id, message_id=msg_id) - except Exception as exc: - logger.debug("[Telegram] could not delete stage emoji in chat=%s: %s", chat_key, exc) + async def _send_running_reply(self, chat_id: str, reply_to_message_id: int) -> None: + # [argus patch #40] Body in _telegram_sender (initial 'received' stage + # emoji + reaction fallback). Kept as a bound method: tests replace it + # per instance, and _send_running_reply_safe dispatches through `self` + # so those overrides keep working. + await _telegram_sender.send_running_reply(self, chat_id, reply_to_message_id) # -- internal ---------------------------------------------------------- @staticmethod @@ -638,10 +606,9 @@ async def _process_incoming_with_reply(self, chat_id: str, msg_id: int, inbound: async def _send_running_reply_safe(self, chat_id: str, msg_id: int) -> None: """Fire-and-forget wrapper for _send_running_reply that logs errors.""" - try: - await self._send_running_reply(chat_id, msg_id) - except Exception: - logger.exception("[Telegram] fire-and-forget _send_running_reply failed in chat=%s", chat_id) + # [argus patch #40] Body in _telegram_sender (dispatches via the bound + # self._send_running_reply, preserving per-instance overrides). + await _telegram_sender.send_running_reply_safe(self, chat_id, msg_id) async def _cmd_generic(self, update, context) -> None: """Forward slash commands to the channel manager.""" diff --git a/backend/tests/test_telegram_sender_seam.py b/backend/tests/test_telegram_sender_seam.py new file mode 100644 index 00000000000..9af7e4654f7 --- /dev/null +++ b/backend/tests/test_telegram_sender_seam.py @@ -0,0 +1,132 @@ +"""[argus patch #40] Seam locks for the extracted Telegram send path. + +telegram.py is upstream-shaped plus three thin shims; the argus send path +lives in app/channels/_telegram_sender.py. These tests turn upstream churn at +the seam into loud failures rather than silent regressions (patch #10's +manager wiring was once lost silently in exactly this way — see PATCHES.md): + + 1. TelegramChannel.send delegates to _telegram_sender.send — a bad 3-way + merge that resurrects upstream's send body (whose superseded stream-edit + helpers are restored as dead code in the same file!) would import fine + and quietly kill the stage-emoji + HTML path. + 2. _send_running_reply delegates likewise, and _send_running_reply_safe + dispatches through the BOUND method (the per-instance override seam that + test_channels.py::TestTelegramProcessingOrder and + test_telegram_channel_connections.py rely on). + 3. OutboundMessage still carries the fields the sender reads. + 4. __init__ still wires _telegram_sender.init_state (state + config parsing). + 5. Channel._on_outbound still dispatches to send() — the bus entry point the + whole extraction hangs off. + +Behavior itself is locked by the pre-existing suite (test_telegram_send.py, +test_channels.py Telegram classes), which this extraction did not modify. +""" + +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, MagicMock + +from app.channels import _telegram_sender +from app.channels.message_bus import MessageBus, OutboundMessage +from app.channels.telegram import TelegramChannel + + +def _run(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +def _channel(config=None): + return TelegramChannel(bus=MessageBus(), config={"bot_token": "t", **(config or {})}) + + +def _msg(**kw): + return OutboundMessage(channel_name="telegram", chat_id="1", thread_id="th", text="x", **kw) + + +class TestShimDelegation: + def test_send_delegates_to_sender_module(self, monkeypatch): + ch = _channel() + ch._application = MagicMock() + mock = AsyncMock() + monkeypatch.setattr(_telegram_sender, "send", mock) + _run(ch.send(_msg(), _max_retries=2)) + mock.assert_awaited_once() + args, kwargs = mock.await_args + assert args[0] is ch + assert args[1].text == "x" + assert kwargs == {"max_retries": 2} + + def test_send_running_reply_delegates_to_sender_module(self, monkeypatch): + ch = _channel() + mock = AsyncMock() + monkeypatch.setattr(_telegram_sender, "send_running_reply", mock) + _run(ch._send_running_reply("1", 5)) + mock.assert_awaited_once_with(ch, "1", 5) + + def test_running_reply_safe_dispatches_through_bound_method(self): + """Per-instance overrides of ch._send_running_reply MUST intercept.""" + ch = _channel() + calls = [] + + async def override(chat_id, msg_id): + calls.append((chat_id, msg_id)) + + ch._send_running_reply = override + _run(ch._send_running_reply_safe("1", 5)) + assert calls == [("1", 5)] + + def test_running_reply_safe_swallows_and_logs_errors(self): + ch = _channel() + + async def boom(chat_id, msg_id): + raise RuntimeError("boom") + + ch._send_running_reply = boom + _run(ch._send_running_reply_safe("1", 5)) # must not raise + + +class TestUpstreamContracts: + def test_outbound_message_has_fields_the_sender_reads(self): + m = _msg(is_final=False, progress_stage="thinking") + assert m.progress_stage == "thinking" + assert m.is_final is False + assert m.chat_id == "1" + + def test_init_wires_sender_state_and_config_parsing(self): + ch = _channel( + { + "stage_min_interval": "2.5", + "working_emoji": "X", + "stage_emoji": {"thinking": "T"}, + } + ) + assert ch._working_msg == {} + assert ch._working_stage == {} + assert ch._working_at == {} + assert ch._promote_timer == {} + assert ch._stage_min_interval == 2.5 + assert ch._stage_emoji["received"] == "X" # working_emoji back-compat + assert ch._stage_emoji["thinking"] == "T" # per-stage override + assert ch._stage_emoji["searching"] == "🔍" # defaults survive overrides + + def test_init_tolerates_empty_config(self): + ch = TelegramChannel(bus=MessageBus(), config={}) + assert ch._stage_min_interval == _telegram_sender._DEFAULT_STAGE_MIN_INTERVAL + assert ch._stage_emoji == _telegram_sender._DEFAULT_STAGE_EMOJI + + def test_send_noops_before_start_sets_application(self): + """Bot is resolved at call time; _application is None until start().""" + ch = _channel() + _run(ch.send(_msg())) # must not raise + + def test_base_on_outbound_dispatches_to_send(self): + ch = _channel() + ch.send = AsyncMock() + m = _msg() + _run(ch._on_outbound(m)) + ch.send.assert_awaited_once_with(m)