From a522a9213fb6558b09e433b916925edc717ca628 Mon Sep 17 00:00:00 2001 From: Maxim Esipov Date: Wed, 5 Aug 2026 13:06:20 +0300 Subject: [PATCH 1/2] fix(gateway): preserve forwarded metadata and profile routing --- gateway/platforms/base.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 6bae26aa052b..5a4c4a5635d4 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2344,6 +2344,9 @@ class MessageEvent: reply_to_author_name: Optional[str] = None reply_to_is_own_message: bool = False # True when the user replied to this bot/assistant's message + # Normalized source metadata for forwarded platform messages. + forward_origin: Optional[Dict[str, str]] = None + # Structured interactive-prompt reply (relay Phase 3). Present when this # event is the user answering a native interactive prompt rendered by the # relay connector (Discord component / Telegram inline keyboard / Slack @@ -6067,6 +6070,9 @@ async def handle_message(self, event: MessageEvent) -> None: if needs_topic_recovery: await asyncio.to_thread(self._apply_topic_recovery, event) + profile = event.source.profile or getattr(self, "_gateway_profile_name", None) + if profile and not event.source.profile: + event.source.profile = profile session_key = build_session_key( event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), From 70504d901b8d3a45e25965871bf6c6bc8cece8e8 Mon Sep 17 00:00:00 2001 From: Maxim Esipov Date: Sat, 8 Aug 2026 09:54:20 +0300 Subject: [PATCH 2/2] fix(telegram): coalesce forwarded startup batches --- gateway/platforms/base.py | 1 + gateway/run.py | 173 +++++ plugins/platforms/telegram/adapter.py | 369 +++++++++- ..._telegram_forwarded_batch_startup_merge.py | 693 ++++++++++++++++++ 4 files changed, 1209 insertions(+), 27 deletions(-) create mode 100644 tests/gateway/test_telegram_forwarded_batch_startup_merge.py diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 5a4c4a5635d4..40c74c52e973 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -2284,6 +2284,7 @@ class MessageType(Enum): AUDIO = "audio" VOICE = "voice" DOCUMENT = "document" + VIDEO_NOTE = "video_note" STICKER = "sticker" COMMAND = "command" # /command style diff --git a/gateway/run.py b/gateway/run.py index 84604b303b7b..fc8486f52166 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -10052,6 +10052,151 @@ def _lookup_session_id_under_store_lock(session_store, session_key: str): # still small enough to never threaten memory. _BUSY_QUEUE_MAX_PENDING = 32 + @staticmethod + def _event_has_batch_media(event: Optional[MessageEvent]) -> bool: + if event is None or not (getattr(event, "media_urls", None) or []): + return False + if getattr(event, "message_type", None) in { + MessageType.PHOTO, + MessageType.VIDEO, + MessageType.AUDIO, + MessageType.VOICE, + MessageType.DOCUMENT, + MessageType.VIDEO_NOTE, + }: + return True + return any( + str(mtype).startswith(("image/", "audio/", "video/", "application/")) + for mtype in (getattr(event, "media_types", None) or []) + ) + + @staticmethod + def _event_has_forwarded_text_context(event: Optional[MessageEvent]) -> bool: + if event is None or getattr(event, "message_type", None) != MessageType.TEXT: + return False + if getattr(event, "media_urls", None): + return False + return bool(getattr(event, "forward_origin", None)) or ( + (getattr(event, "text", None) or "").lstrip().startswith("[Forwarded message |") + ) + + @classmethod + def _event_can_join_startup_batch(cls, event: Optional[MessageEvent]) -> bool: + return cls._event_has_batch_media(event) or cls._event_has_forwarded_text_context(event) + + @staticmethod + def _adapter_declared_method(adapter: Any, name: str) -> Optional[Callable[..., Any]]: + try: + inspect.getattr_static(adapter, name) + except AttributeError: + return None + method = getattr(adapter, name, None) + return method if callable(method) else None + + @staticmethod + def _startup_media_grace_seconds() -> float: + return 1.0 + + @staticmethod + def _format_forward_origin_context(forward_origin: Optional[Dict[str, str]]) -> Optional[str]: + if not forward_origin: + return None + parts = ["Forwarded message"] + if forward_origin.get("automatic") == "true": + parts.append("automatic forward") + sender = forward_origin.get("sender_name") + if sender: + username = forward_origin.get("sender_username") + parts.append(f"From: {sender} (@{username})" if username else f"From: {sender}") + elif forward_origin.get("type") == "hidden_user": + parts.append("From: hidden sender") + chat = forward_origin.get("chat_name") + if chat: + username = forward_origin.get("chat_username") + parts.append(f"Chat: {chat} (@{username})" if username else f"Chat: {chat}") + if forward_origin.get("author_signature"): + parts.append(f"Author: {forward_origin['author_signature']}") + if forward_origin.get("date"): + parts.append(f"Date: {forward_origin['date']}") + return "[" + " | ".join(parts) + "]" + + def _inline_forward_context(self, event: MessageEvent) -> MessageEvent: + if not getattr(event, "forward_origin", None): + return event + context = self._format_forward_origin_context(event.forward_origin) + if not context: + return event + text = event.text or "" + if text.lstrip().startswith("[Forwarded message |"): + return dataclasses.replace(event, forward_origin=None) + return dataclasses.replace( + event, + text=f"{context}\n\n{text}" if text else context, + forward_origin=None, + ) + + async def _merge_startup_media_followups( + self, + event: MessageEvent, + source: SessionSource, + session_key: str, + ) -> MessageEvent: + """Merge a rapid Telegram forward batch into its starting text turn.""" + if ( + source.platform != Platform.TELEGRAM + or event.message_type != MessageType.TEXT + or getattr(event, "media_urls", None) + ): + return event + + adapter = self._adapter_for_source(source) + if adapter is None: + return event + pop_media = self._adapter_declared_method(adapter, "pop_startup_media_event") + has_pending = self._adapter_declared_method(adapter, "has_startup_media_pending") + loop = asyncio.get_running_loop() + deadline = loop.time() + self._startup_media_grace_seconds() + merged_attachments = 0 + merged_forwarded_texts = 0 + + while True: + incoming = None + if pop_media is not None: + try: + incoming = pop_media(session_key) + except Exception: + logger.debug("Telegram startup media pop failed", exc_info=True) + incoming = None + if incoming is not None: + if getattr(incoming, "forward_origin", None): + incoming = self._inline_forward_context(incoming) + if incoming.message_type == MessageType.TEXT and not incoming.media_urls: + merged_forwarded_texts += 1 + slot = {session_key: event} + merge_pending_message_event(slot, session_key, incoming, merge_text=True) + event = slot[session_key] + merged_attachments += len(getattr(incoming, "media_urls", None) or []) + continue + + pending = False + if has_pending is not None: + try: + pending = bool(has_pending(session_key)) + except Exception: + logger.debug("Telegram startup pending check failed", exc_info=True) + if not pending or loop.time() >= deadline: + break + await asyncio.sleep(min(0.05, max(0.0, deadline - loop.time()))) + + if merged_attachments or merged_forwarded_texts: + logger.info( + "Merged %d Telegram startup attachment(s) and %d forwarded text batch(es) into session %s", + merged_attachments, + merged_forwarded_texts, + session_key, + ) + return event + def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) -> None: adapter = self._adapter_for_source(event.source) if not adapter: @@ -10292,6 +10437,28 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session _busy_state = self._peek_session_state(session_key) running_agent = _busy_state.turn.agent if _busy_state else None + if ( + event.source.platform == Platform.TELEGRAM + and running_agent is _AGENT_PENDING_SENTINEL + and self._event_can_join_startup_batch(event) + ): + logger.debug( + "Queueing Telegram startup forward/media follow-up for session %s without interrupt/ack", + session_key, + ) + queue_startup = self._adapter_declared_method(adapter, "queue_startup_batch_event") + if queue_startup is not None: + queue_startup(session_key, event) + else: + merge_pending_message_event( + adapter._pending_messages, + session_key, + event, + merge_text=self._event_has_forwarded_text_context(event), + ) + return True + + effective_mode = self._effective_busy_input_mode(event.source) busy_text_mode = self._effective_busy_text_mode(event.source) if ( event.message_type == MessageType.TEXT @@ -15393,6 +15560,7 @@ def _configure_profile_adapter( # profile-scoped. Preserve both dimensions in the key so dashboard # and NAS health aggregation can see which secondary profile failed. adapter._runtime_status_platform_key = f"{profile_name}:{platform.value}" + adapter._gateway_profile_name = profile_name adapter.set_message_handler(self._make_profile_message_handler(profile_name)) adapter.set_fatal_error_handler( self._make_profile_fatal_error_handler(profile_name, platform) @@ -18151,6 +18319,10 @@ async def _prepare_inbound_message_text( if getattr(event, "channel_context", None): message_text = f"{event.channel_context}\n\n[New message]\n{message_text}" + forward_context = self._format_forward_origin_context(getattr(event, "forward_origin", None)) + if forward_context: + message_text = f"{forward_context}\n\n{message_text}" + # Declare at outer scope so the audio-file-paths handling block below # remains safe when ``event.media_urls`` is empty (no inner block runs). audio_file_paths: list[str] = [] @@ -20077,6 +20249,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g # attachments (documents, audio, etc.) are not sent to the vision # tool even when they appear in the same message. # ----------------------------------------------------------------- + event = await self._merge_startup_media_followups(event, source, session_key) message_text = await self._prepare_profile_scoped_inbound_message_text( event=event, source=source, diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 7fb1469bbdc9..09a0e2169e96 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/plugins/platforms/telegram/adapter.py @@ -287,6 +287,7 @@ class _MockContextTypes: MessageType, ProcessingOutcome, SendResult, + merge_pending_message_event, classify_send_error, cache_image_from_bytes, cache_audio_from_bytes, @@ -646,6 +647,20 @@ class TelegramAdapter(BasePlatformAdapter): - Media messages """ + @staticmethod + def _media_message_filter(): + """Return all supported PTB media filters, including video notes.""" + media_filter = ( + filters.PHOTO + | filters.VIDEO + | filters.AUDIO + | filters.VOICE + | filters.Document.ALL + | filters.Sticker.ALL + ) + video_note_filter = getattr(filters, "VIDEO_NOTE", None) + return media_filter | video_note_filter if video_note_filter is not None else media_filter + # Telegram message limits MAX_MESSAGE_LENGTH = 4096 supports_code_blocks = True # Telegram MarkdownV2 renders fenced code blocks @@ -773,6 +788,9 @@ def __init__(self, config: PlatformConfig): self._pending_photo_batch_tasks: Dict[str, asyncio.Task] = {} self._media_group_events: Dict[str, MessageEvent] = {} self._media_group_tasks: Dict[str, asyncio.Task] = {} + self._media_downloads_in_progress_by_session: Dict[str, int] = {} + # Dedicated startup coalescing slot, separate from the normal FIFO. + self._startup_batch_events: Dict[str, MessageEvent] = {} # Buffer rapid text messages so Telegram client-side splits of long # messages are aggregated into a single MessageEvent. Lower defaults # (0.3s / 1.0s instead of 0.6s / 2.0s) let short replies stream @@ -4909,6 +4927,8 @@ def collect(task: Optional[asyncio.Task]) -> None: self._pending_photo_batches.clear() self._pending_text_batch_tasks.clear() self._pending_text_batches.clear() + getattr(self, "_startup_batch_events", {}).clear() + getattr(self, "_media_downloads_in_progress_by_session", {}).clear() if getattr(self, "_polling_error_task", None) is not current_task: self._polling_error_task = None if getattr(self, "_polling_progress_verifier_task", None) is not current_task: @@ -9155,6 +9175,9 @@ def _media_message_type(self, msg: Message) -> MessageType: return MessageType.STICKER if msg.photo: return MessageType.PHOTO + video_note = getattr(msg, "video_note", None) + if video_note is not None and type(video_note).__module__.startswith("telegram"): + return MessageType.VIDEO_NOTE if msg.video: return MessageType.VIDEO if msg.audio: @@ -9632,15 +9655,47 @@ async def _handle_location_message(self, update: Update, context: ContextTypes.D # Text message aggregation (handles Telegram client-side splits) # ------------------------------------------------------------------ - def _text_batch_key(self, event: MessageEvent) -> str: - """Session-scoped key for text message batching. + @staticmethod + def _format_forward_origin_context( + forward_origin: Optional[Dict[str, str]], + ) -> Optional[str]: + if not forward_origin: + return None + parts = ["Forwarded message"] + if forward_origin.get("automatic") == "true": + parts.append("automatic forward") + sender = forward_origin.get("sender_name") + if sender: + username = forward_origin.get("sender_username") + parts.append(f"From: {sender} (@{username})" if username else f"From: {sender}") + elif forward_origin.get("type") == "hidden_user": + parts.append("From: hidden sender") + chat = forward_origin.get("chat_name") + if chat: + username = forward_origin.get("chat_username") + parts.append(f"Chat: {chat} (@{username})" if username else f"Chat: {chat}") + if forward_origin.get("author_signature"): + parts.append(f"Author: {forward_origin['author_signature']}") + if forward_origin.get("date"): + parts.append(f"Date: {forward_origin['date']}") + return "[" + " | ".join(parts) + "]" + + def _event_with_inline_forward_context(self, event: MessageEvent) -> MessageEvent: + context = self._format_forward_origin_context(event.forward_origin) + if not context: + return event + if not (event.text or "").lstrip().startswith("[Forwarded message |"): + event.text = f"{context}\n\n{event.text}" if event.text else context + event.forward_origin = None + return event - Applies the installed topic-recovery hook first so DM-topic batches - coalesce on (and dispatch to) the recovered lane rather than the - raw inbound ``message_thread_id`` Telegram may have attached. - """ + def _event_session_key(self, event: MessageEvent) -> str: from gateway.session import build_session_key + self._apply_topic_recovery(event) + profile = event.source.profile or getattr(self, "_gateway_profile_name", None) + if profile and not event.source.profile: + event.source.profile = profile return build_session_key( event.source, group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), @@ -9648,6 +9703,106 @@ def _text_batch_key(self, event: MessageEvent) -> str: profile=self._session_key_profile(event.source), ) + def _track_media_download_start(self, event: MessageEvent) -> str: + session_key = self._event_session_key(event) + counts = self._media_downloads_in_progress_by_session + counts[session_key] = counts.get(session_key, 0) + 1 + return session_key + + def _track_media_download_done(self, session_key: Optional[str]) -> None: + if not session_key: + return + remaining = self._media_downloads_in_progress_by_session.get(session_key, 0) - 1 + if remaining > 0: + self._media_downloads_in_progress_by_session[session_key] = remaining + else: + self._media_downloads_in_progress_by_session.pop(session_key, None) + + def queue_startup_batch_event(self, session_key: str, event: MessageEvent) -> None: + event = self._event_with_inline_forward_context(event) + merge_pending_message_event( + self._startup_batch_events, + session_key, + event, + merge_text=(event.message_type == MessageType.TEXT), + ) + + def has_startup_media_pending(self, session_key: str) -> bool: + if not session_key: + return False + if session_key in self._startup_batch_events: + return True + if self._media_downloads_in_progress_by_session.get(session_key, 0) > 0: + return True + prefix = f"{session_key}:" + if any( + key == f"{session_key}:photo-burst" or key.startswith(prefix + "album:") + for key in self._pending_photo_batches + ): + return True + for event in self._media_group_events.values(): + try: + if self._event_session_key(event) == session_key: + return True + except Exception: + continue + forwarded_text = self._pending_text_batches.get(session_key) + return bool( + forwarded_text + and (forwarded_text.text or "").lstrip().startswith("[Forwarded message |") + ) + + def pop_startup_media_event(self, session_key: str) -> Optional[MessageEvent]: + if not session_key: + return None + merged: Dict[str, MessageEvent] = {} + startup_event = self._startup_batch_events.pop(session_key, None) + if startup_event is not None: + merge_pending_message_event( + merged, + session_key, + startup_event, + merge_text=(startup_event.message_type == MessageType.TEXT), + ) + text_event = self._pending_text_batches.get(session_key) + if text_event and (text_event.text or "").lstrip().startswith("[Forwarded message |"): + text_event = self._pending_text_batches.pop(session_key, None) + task = self._pending_text_batch_tasks.pop(session_key, None) + if task is not None and not task.done(): + task.cancel() + if text_event is not None: + merge_pending_message_event(merged, session_key, text_event, merge_text=True) + photo_keys = [ + key for key in list(self._pending_photo_batches) + if key == f"{session_key}:photo-burst" or key.startswith(f"{session_key}:album:") + ] + for key in photo_keys: + event = self._pending_photo_batches.pop(key, None) + task = self._pending_photo_batch_tasks.pop(key, None) + if task is not None and not task.done(): + task.cancel() + if event is not None and event.media_urls: + merge_pending_message_event(merged, session_key, event) + group_ids = [] + for group_id, event in list(self._media_group_events.items()): + try: + if self._event_session_key(event) == session_key: + group_ids.append(group_id) + except Exception: + continue + for group_id in group_ids: + event = self._media_group_events.pop(group_id, None) + task = self._media_group_tasks.pop(group_id, None) + if task is not None and not task.done(): + task.cancel() + if event is not None and event.media_urls: + merge_pending_message_event(merged, session_key, event) + return merged.get(session_key) + + def _text_batch_key(self, event: MessageEvent) -> str: + """Session-scoped key for text message batching.""" + return self._event_session_key(event) + def _enqueue_text_event(self, event: MessageEvent) -> None: """Buffer a text event and reset the flush timer. @@ -9660,6 +9815,7 @@ def _enqueue_text_event(self, event: MessageEvent) -> None: self._hold_inbound_event(event, where="text-enqueue") return + event = self._event_with_inline_forward_context(event) key = self._text_batch_key(event) existing = self._pending_text_batches.get(key) chunk_len = len(event.text or "") @@ -9745,13 +9901,7 @@ async def _flush_text_batch(self, key: str) -> None: def _photo_batch_key(self, event: MessageEvent, msg: Message) -> str: """Return a batching key for Telegram photos/albums.""" - from gateway.session import build_session_key - session_key = build_session_key( - event.source, - group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True), - thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False), - profile=self._session_key_profile(event.source), - ) + session_key = self._event_session_key(event) media_group_id = getattr(msg, "media_group_id", None) if media_group_id: return f"{session_key}:album:{media_group_id}" @@ -9789,12 +9939,13 @@ def _enqueue_photo_event(self, batch_key: str, event: MessageEvent) -> None: existing = self._pending_photo_batches.get(batch_key) if existing is None: - self._pending_photo_batches[batch_key] = event + self._pending_photo_batches[batch_key] = self._event_with_inline_forward_context(event) else: - existing.media_urls.extend(event.media_urls) - existing.media_types.extend(event.media_types) - if event.text: - existing.text = self._merge_caption(existing.text, event.text) + merge_pending_message_event( + self._pending_photo_batches, + batch_key, + self._event_with_inline_forward_context(event), + ) prior_task = self._pending_photo_batch_tasks.get(batch_key) if prior_task and not prior_task.done(): @@ -9850,6 +10001,7 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA # Download photo to local image cache so the vision tool can access it # even after Telegram's ephemeral file URLs expire (~1 hour). if msg.photo: + download_session_key = self._track_media_download_start(event) try: # msg.photo is a list of PhotoSize sorted by size; take the largest photo = msg.photo[-1] @@ -9879,9 +10031,17 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA except Exception as e: logger.warning("[Telegram] Failed to cache photo: %s", _redact_telegram_error_text(e), exc_info=True) await self._surface_media_cache_failure(msg, event, "photo", e) + finally: + self._track_media_download_done(download_session_key) + + # Keep non-photo download tracking alive through BasePlatformAdapter + # dispatch. That path may await topic recovery before the busy/startup + # handler registers the event; clearing earlier creates a false idle gap. + dispatch_download_session_key = None # Download voice/audio messages to cache for STT transcription if msg.voice: + download_session_key = None try: allowed, note = self._telegram_media_size_allowed(msg.voice, "voice message") if not allowed: @@ -9889,16 +10049,25 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA logger.info("[Telegram] Skipped oversized user voice (size=%s)", getattr(msg.voice, "file_size", None)) await self.handle_message(event) return + download_session_key = self._track_media_download_start(event) file_obj = await msg.voice.get_file() audio_bytes = await file_obj.download_as_bytearray() cached_path = cache_audio_from_bytes(bytes(audio_bytes), ext=".ogg") event.media_urls = [cached_path] event.media_types = ["audio/ogg"] logger.info("[Telegram] Cached user voice at %s", cached_path) + dispatch_download_session_key = download_session_key + download_session_key = None except Exception as e: logger.warning("[Telegram] Failed to cache voice: %s", _redact_telegram_error_text(e), exc_info=True) await self._surface_media_cache_failure(msg, event, "voice message", e) + if download_session_key: + dispatch_download_session_key = download_session_key + download_session_key = None + finally: + self._track_media_download_done(download_session_key) elif msg.audio: + download_session_key = None try: allowed, note = self._telegram_media_size_allowed(msg.audio, "audio file") if not allowed: @@ -9906,17 +10075,63 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA logger.info("[Telegram] Skipped oversized user audio (size=%s)", getattr(msg.audio, "file_size", None)) await self.handle_message(event) return + download_session_key = self._track_media_download_start(event) file_obj = await msg.audio.get_file() audio_bytes = await file_obj.download_as_bytearray() cached_path = cache_audio_from_bytes(bytes(audio_bytes), ext=".mp3") event.media_urls = [cached_path] event.media_types = ["audio/mp3"] logger.info("[Telegram] Cached user audio at %s", cached_path) + dispatch_download_session_key = download_session_key + download_session_key = None except Exception as e: logger.warning("[Telegram] Failed to cache audio: %s", _redact_telegram_error_text(e), exc_info=True) await self._surface_media_cache_failure(msg, event, "audio file", e) + if download_session_key: + dispatch_download_session_key = download_session_key + download_session_key = None + finally: + self._track_media_download_done(download_session_key) + + elif ( + (video_note := getattr(msg, "video_note", None)) is not None + and type(video_note).__module__.startswith("telegram") + ): + download_session_key = None + try: + allowed, note = self._telegram_media_size_allowed(msg.video_note, "video note") + if not allowed: + event.text = self._append_observed_note(event.text, note or "") + logger.info( + "[Telegram] Skipped oversized user video note (size=%s)", + getattr(msg.video_note, "file_size", None), + ) + await self.handle_message(event) + return + download_session_key = self._track_media_download_start(event) + file_obj = await msg.video_note.get_file() + video_bytes = await file_obj.download_as_bytearray() + cached_path = cache_video_from_bytes(bytes(video_bytes), ext=".mp4") + event.media_urls = [cached_path] + event.media_types = ["video/mp4"] + logger.info("[Telegram] Cached user video note at %s", cached_path) + dispatch_download_session_key = download_session_key + download_session_key = None + except Exception as e: + logger.warning( + "[Telegram] Failed to cache video note: %s", + _redact_telegram_error_text(e), + exc_info=True, + ) + await self._surface_media_cache_failure(msg, event, "video note", e) + if download_session_key: + dispatch_download_session_key = download_session_key + download_session_key = None + finally: + self._track_media_download_done(download_session_key) elif msg.video: + download_session_key = None try: allowed, note = self._telegram_media_size_allowed(msg.video, "video file") if not allowed: @@ -9924,6 +10139,7 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA logger.info("[Telegram] Skipped oversized user video (size=%s)", getattr(msg.video, "file_size", None)) await self.handle_message(event) return + download_session_key = self._track_media_download_start(event) file_obj = await msg.video.get_file() video_bytes = await file_obj.download_as_bytearray() ext = ".mp4" @@ -9936,13 +10152,21 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA event.media_urls = [cached_path] event.media_types = [SUPPORTED_VIDEO_TYPES.get(ext, "video/mp4")] logger.info("[Telegram] Cached user video at %s", cached_path) + dispatch_download_session_key = download_session_key + download_session_key = None except Exception as e: logger.warning("[Telegram] Failed to cache video: %s", _redact_telegram_error_text(e), exc_info=True) await self._surface_media_cache_failure(msg, event, "video file", e) + if download_session_key: + dispatch_download_session_key = download_session_key + download_session_key = None + finally: + self._track_media_download_done(download_session_key) # Download document files to cache for agent processing elif msg.document: doc = msg.document + download_session_key = None try: # Determine file extension ext = "" @@ -9974,6 +10198,8 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA await self.handle_message(event) return + download_session_key = self._track_media_download_start(event) + # Telegram may deliver screenshots/photos as documents. If the # payload is actually an image, route it through the image cache # and batching path instead of rejecting it as a document. @@ -10088,19 +10314,30 @@ async def _handle_media_message(self, update: Update, context: ContextTypes.DEFA # terminal/read_file against it. No inline injection. pass + dispatch_download_session_key = download_session_key + download_session_key = None + except Exception as e: logger.warning("[Telegram] Failed to cache document: %s", _redact_telegram_error_text(e), exc_info=True) await self._surface_media_cache_failure( msg, event, "attachment", e, display_name=getattr(doc, "file_name", None) or None, ) + if download_session_key: + dispatch_download_session_key = download_session_key + download_session_key = None + finally: + self._track_media_download_done(download_session_key) - media_group_id = getattr(msg, "media_group_id", None) - if media_group_id: - await self._queue_media_group_event(str(media_group_id), event) - return + try: + media_group_id = getattr(msg, "media_group_id", None) + if media_group_id: + await self._queue_media_group_event(str(media_group_id), event) + return - await self.handle_message(event) + await self.handle_message(event) + finally: + self._track_media_download_done(dispatch_download_session_key) async def _queue_media_group_event(self, media_group_id: str, event: MessageEvent) -> None: """Buffer Telegram media-group items so albums arrive as one logical event. @@ -10114,14 +10351,16 @@ async def _queue_media_group_event(self, media_group_id: str, event: MessageEven self._hold_inbound_event(event, where="media-group-enqueue") return + event = self._event_with_inline_forward_context(event) existing = self._media_group_events.get(media_group_id) if existing is None: self._media_group_events[media_group_id] = event else: - existing.media_urls.extend(event.media_urls) - existing.media_types.extend(event.media_types) - if event.text: - existing.text = self._merge_caption(existing.text, event.text) + merge_pending_message_event( + self._media_group_events, + media_group_id, + event, + ) prior_task = self._media_group_tasks.get(media_group_id) if prior_task: @@ -10136,6 +10375,18 @@ async def _flush_media_group_event(self, media_group_id: str) -> None: event = None try: await asyncio.sleep(self.MEDIA_GROUP_WAIT_SECONDS) + deadline = asyncio.get_running_loop().time() + 3.0 + while True: + pending = self._media_group_events.get(media_group_id) + if pending is None: + break + session_key = self._event_session_key(pending) + downloads = getattr(self, "_media_downloads_in_progress_by_session", {}) + if downloads.get(session_key, 0) <= 0: + break + if asyncio.get_running_loop().time() >= deadline: + break + await asyncio.sleep(0.05) event = self._media_group_events.pop(media_group_id, None) if event is None: return @@ -10387,6 +10638,69 @@ def _extract_rich_reply_text(cls, reply_to_message: Any) -> Optional[str]: except Exception: return None + @staticmethod + def _telegram_forward_origin_type(origin: Any) -> str: + origin_type = getattr(origin, "type", None) + if origin_type is None: + return "unknown" + return str(getattr(origin_type, "name", origin_type) or "unknown").lower() + + @staticmethod + def _telegram_forward_origin_date(origin: Any) -> Optional[str]: + date = getattr(origin, "date", None) + if date is None: + return None + if hasattr(date, "isoformat"): + return date.isoformat() + return str(date) + + def _extract_forward_origin(self, message: Message) -> Optional[Dict[str, str]]: + """Normalize Telegram forwarded-message metadata for agent context.""" + origin = getattr(message, "forward_origin", None) + if origin is None: + return None + origin_type = getattr(origin, "type", None) + if not isinstance(origin_type, str): + return None + result: Dict[str, str] = {"type": self._telegram_forward_origin_type(origin)} + if getattr(message, "is_automatic_forward", False): + result["automatic"] = "true" + date = self._telegram_forward_origin_date(origin) + if date: + result["date"] = date + sender_user = getattr(origin, "sender_user", None) + if sender_user is not None: + sender_name = getattr(sender_user, "full_name", None) or getattr(sender_user, "username", None) + if sender_name: + result["sender_name"] = str(sender_name) + sender_id = getattr(sender_user, "id", None) + if sender_id is not None: + result["sender_id"] = str(sender_id) + username = getattr(sender_user, "username", None) + if username: + result["sender_username"] = str(username) + hidden_name = getattr(origin, "sender_user_name", None) + if hidden_name: + result["sender_name"] = str(hidden_name) + chat = getattr(origin, "chat", None) + if chat is not None: + chat_name = getattr(chat, "title", None) or getattr(chat, "full_name", None) or getattr(chat, "username", None) + if chat_name: + result["chat_name"] = str(chat_name) + chat_id = getattr(chat, "id", None) + if chat_id is not None: + result["chat_id"] = str(chat_id) + username = getattr(chat, "username", None) + if username: + result["chat_username"] = str(username) + author_signature = getattr(origin, "author_signature", None) + if author_signature: + result["author_signature"] = str(author_signature) + message_id = getattr(origin, "message_id", None) + if message_id is not None: + result["message_id"] = str(message_id) + return result + def _build_message_event( self, message: Message, @@ -10550,6 +10864,7 @@ def _build_message_event( platform_update_id=update_id, reply_to_message_id=reply_to_id, reply_to_text=reply_to_text, + forward_origin=self._extract_forward_origin(message), auto_skill=topic_skill, channel_prompt=_channel_prompt, timestamp=message.date, diff --git a/tests/gateway/test_telegram_forwarded_batch_startup_merge.py b/tests/gateway/test_telegram_forwarded_batch_startup_merge.py new file mode 100644 index 000000000000..30fe735c1658 --- /dev/null +++ b/tests/gateway/test_telegram_forwarded_batch_startup_merge.py @@ -0,0 +1,693 @@ +import asyncio +from datetime import datetime, timezone +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType +from plugins.platforms.telegram.adapter import TelegramAdapter +from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL +from gateway.session import SessionSource, build_session_key + + +class _DummyTask: + def __init__(self): + self.cancelled = False + + def done(self): + return False + + def cancel(self): + self.cancelled = True + + +def _source() -> SessionSource: + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="273403055", + chat_type="dm", + user_id="273403055", + user_name="Maxim E.", + thread_id="363402", + ) + + +def _text_event(source: SessionSource) -> MessageEvent: + return MessageEvent( + text="Is this a real study?", + message_type=MessageType.TEXT, + source=source, + ) + + +def _photo_event(source: SessionSource, path: str = "/tmp/alcohol-study.jpg") -> MessageEvent: + return MessageEvent( + text="", + message_type=MessageType.PHOTO, + source=source, + media_urls=[path], + media_types=["image/jpeg"], + ) + + +def _forwarded_photo_event( + source: SessionSource, + path: str = "/tmp/forwarded-post.jpg", +) -> MessageEvent: + return MessageEvent( + text="Forwarded post body", + message_type=MessageType.PHOTO, + source=source, + media_urls=[path], + media_types=["image/jpeg"], + forward_origin={ + "type": "channel", + "chat_name": "AI Channel", + "chat_username": "ai_channel", + "date": "2026-07-28T09:02:43+00:00", + }, + ) + + +def _document_event( + source: SessionSource, + path: str = "/root/.hermes/cache/documents/doc_abcd_guide.docx", +) -> MessageEvent: + return MessageEvent( + text="", + message_type=MessageType.DOCUMENT, + source=source, + media_urls=[path], + media_types=["application/vnd.openxmlformats-officedocument.wordprocessingml.document"], + forward_origin={"type": "user", "sender_name": "Alice"}, + ) + + +def _forwarded_text_event(source: SessionSource) -> MessageEvent: + return MessageEvent( + text="sk-or-v1-example\nsecond forwarded text", + message_type=MessageType.TEXT, + source=source, + forward_origin={ + "type": "user", + "sender_name": "Alina", + "date": "2026-06-14T21:03:26+00:00", + }, + ) + + +def _make_adapter() -> TelegramAdapter: + adapter = TelegramAdapter.__new__(TelegramAdapter) + adapter.config = PlatformConfig(enabled=True, token="fake") + adapter._pending_messages = {} + adapter._pending_photo_batches = {} + adapter._pending_photo_batch_tasks = {} + adapter._media_group_events = {} + adapter._media_group_tasks = {} + adapter._media_downloads_in_progress_by_session = {} + adapter._startup_batch_events = {} + adapter._pending_text_batches = {} + adapter._pending_text_batch_tasks = {} + adapter._drop_delayed_deliveries = False + adapter._text_batch_delay_seconds = 0.3 + adapter._text_batch_split_delay_seconds = 1.0 + adapter._apply_topic_recovery = lambda _event: None + return adapter + + +def _make_runner(adapter: TelegramAdapter) -> GatewayRunner: + runner = GatewayRunner.__new__(GatewayRunner) + runner.config = GatewayConfig( + platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake")}, + ) + runner.adapters = {Platform.TELEGRAM: adapter} + runner._pending_native_image_paths_by_session = {} + runner._model = "openai/gpt-4.1-mini" + runner._base_url = None + runner._decide_image_input_mode = lambda **_kwargs: "native" + runner._is_user_authorized = lambda _source: True + return runner + + +def test_media_filter_includes_video_note(monkeypatch): + class _Filter: + def __init__(self, names): + self.names = set(names) + + def __or__(self, other): + return _Filter(self.names | other.names) + + fake_filters = SimpleNamespace( + PHOTO=_Filter({"photo"}), + VIDEO=_Filter({"video"}), + AUDIO=_Filter({"audio"}), + VOICE=_Filter({"voice"}), + VIDEO_NOTE=_Filter({"video_note"}), + Document=SimpleNamespace(ALL=_Filter({"document"})), + Sticker=SimpleNamespace(ALL=_Filter({"sticker"})), + ) + monkeypatch.setattr("plugins.platforms.telegram.adapter.filters", fake_filters) + + media_filter = TelegramAdapter._media_message_filter() + + assert "video_note" in media_filter.names + + +def test_telegram_forward_origin_extraction_user_origin(): + adapter = _make_adapter() + message = SimpleNamespace( + is_automatic_forward=False, + forward_origin=SimpleNamespace( + type="user", + date=datetime(2026, 6, 14, 9, 0, tzinfo=timezone.utc), + sender_user=SimpleNamespace( + id=42, + full_name="Skippy", + username="skippy_bot", + ), + ), + ) + + assert adapter._extract_forward_origin(message) == { + "type": "user", + "date": "2026-06-14T09:00:00+00:00", + "sender_name": "Skippy", + "sender_id": "42", + "sender_username": "skippy_bot", + } + + +def test_telegram_forward_origin_extraction_channel_origin(): + adapter = _make_adapter() + message = SimpleNamespace( + is_automatic_forward=False, + forward_origin=SimpleNamespace( + type="channel", + date=datetime(2026, 7, 28, 9, 2, 43, tzinfo=timezone.utc), + chat=SimpleNamespace( + id=-1003091706822, + title="AI Channel", + full_name=None, + username="ai_channel", + ), + author_signature="Editor", + message_id=382, + ), + ) + + assert adapter._extract_forward_origin(message) == { + "type": "channel", + "date": "2026-07-28T09:02:43+00:00", + "chat_name": "AI Channel", + "chat_id": "-1003091706822", + "chat_username": "ai_channel", + "author_signature": "Editor", + "message_id": "382", + } + + +@pytest.mark.asyncio +async def test_multiple_forwarded_texts_preserve_each_origin_in_order(): + source = _source() + adapter = _make_adapter() + first = MessageEvent( + text="first", + message_type=MessageType.TEXT, + source=source, + forward_origin={"type": "user", "sender_name": "Alice"}, + ) + second = MessageEvent( + text="second", + message_type=MessageType.TEXT, + source=source, + forward_origin={"type": "user", "sender_name": "Bob"}, + ) + + adapter._enqueue_text_event(first) + adapter._enqueue_text_event(second) + batch = adapter._pending_text_batches[adapter._text_batch_key(first)] + + assert batch.forward_origin is None + assert batch.text == ( + "[Forwarded message | From: Alice]\n\nfirst\n" + "[Forwarded message | From: Bob]\n\nsecond" + ) + for task in adapter._pending_text_batch_tasks.values(): + task.cancel() + + +@pytest.mark.asyncio +async def test_gateway_merges_buffered_photo_batch_before_image_routing(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + task = _DummyTask() + adapter._pending_photo_batches[f"{session_key}:photo-burst"] = _photo_event(source) + adapter._pending_photo_batch_tasks[f"{session_key}:photo-burst"] = task + + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + message_text = await runner._prepare_inbound_message_text( + event=event, + source=source, + history=[], + ) + + assert message_text == "Is this a real study?" + assert event.message_type == MessageType.PHOTO + assert event.media_urls == ["/tmp/alcohol-study.jpg"] + assert runner._consume_pending_native_image_paths(session_key) == ["/tmp/alcohol-study.jpg"] + assert adapter._pending_photo_batches == {} + assert task.cancelled is True + + +@pytest.mark.asyncio +async def test_forwarded_photo_keeps_origin_on_forwarded_payload_not_user_question(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + adapter._pending_photo_batches[f"{session_key}:photo-burst"] = _forwarded_photo_event(source) + + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + message_text = await runner._prepare_inbound_message_text( + event=event, + source=source, + history=[], + ) + + assert event.forward_origin is None + assert message_text.startswith("Is this a real study?") + assert ( + "[Forwarded message | Chat: AI Channel (@ai_channel) | " + "Date: 2026-07-28T09:02:43+00:00]" + ) in message_text + assert message_text.index("Is this a real study?") < message_text.index("[Forwarded message |") + assert message_text.endswith("Forwarded post body") + assert event.media_urls == ["/tmp/forwarded-post.jpg"] + + +@pytest.mark.asyncio +async def test_photo_burst_preserves_late_forward_origin_inline(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + key = f"{session_key}:photo-burst" + + adapter._enqueue_photo_event(key, _photo_event(source, "/tmp/plain.jpg")) + adapter._enqueue_photo_event(key, _forwarded_photo_event(source, "/tmp/forwarded.jpg")) + + batch = adapter._pending_photo_batches[key] + assert batch.forward_origin is None + assert batch.media_urls == ["/tmp/plain.jpg", "/tmp/forwarded.jpg"] + assert "[Forwarded message | Chat: AI Channel (@ai_channel)" in batch.text + adapter._pending_photo_batch_tasks[key].cancel() + + +@pytest.mark.asyncio +async def test_album_flush_waits_for_other_downloads(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + adapter.MEDIA_GROUP_WAIT_SECONDS = 0 + adapter.handle_message = AsyncMock() + adapter._media_group_events["album-1"] = _photo_event(source) + adapter._media_downloads_in_progress_by_session[session_key] = 1 + + async def finish_second_download(): + await asyncio.sleep(0.02) + adapter._media_downloads_in_progress_by_session.pop(session_key, None) + + producer = asyncio.create_task(finish_second_download()) + await adapter._flush_media_group_event("album-1") + await producer + + adapter.handle_message.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_gateway_waits_for_in_progress_photo_download(monkeypatch): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + adapter._media_downloads_in_progress_by_session[session_key] = 1 + monkeypatch.setattr(runner, "_startup_media_grace_seconds", lambda: 0.2) + + async def finish_download(): + await asyncio.sleep(0.02) + adapter._pending_photo_batches[f"{session_key}:photo-burst"] = _photo_event( + source, + "/tmp/late-photo.jpg", + ) + adapter._media_downloads_in_progress_by_session.pop(session_key, None) + + producer = asyncio.create_task(finish_download()) + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + await producer + + assert event.message_type == MessageType.PHOTO + assert event.text == "Is this a real study?" + assert event.media_urls == ["/tmp/late-photo.jpg"] + assert adapter._pending_photo_batches == {} + + +@pytest.mark.asyncio +async def test_non_photo_tracker_stays_live_until_startup_registration(monkeypatch): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + runner._running_agents = {session_key: _AGENT_PENDING_SENTINEL} + runner._busy_input_mode = "interrupt" + runner._busy_text_mode = "interrupt" + runner._busy_ack_ts = {} + runner._running_agents_ts = {} + monkeypatch.setattr(runner, "_startup_media_grace_seconds", lambda: 0.2) + + adapter._is_user_authorized_from_message = lambda _message: True + adapter._should_process_message = lambda _message: True + adapter._media_message_type = lambda _message: MessageType.VOICE + adapter._build_message_event = lambda _message, _type, update_id=None: MessageEvent( + text="", + message_type=MessageType.VOICE, + source=source, + platform_update_id=update_id, + ) + adapter._apply_telegram_group_observe_attribution = lambda event: event + + async def delayed_registration(event): + await asyncio.sleep(0.05) + handled = await runner._handle_active_session_busy_message(event, session_key) + assert handled is True + + adapter.handle_message = delayed_registration + + class _File: + async def download_as_bytearray(self): + await asyncio.sleep(0.01) + return bytearray(b"voice") + + class _Voice: + file_size = 5 + + async def get_file(self): + return _File() + + msg = SimpleNamespace( + caption=None, + sticker=None, + photo=None, + voice=_Voice(), + audio=None, + video_note=None, + video=None, + document=None, + media_group_id=None, + ) + update = SimpleNamespace(message=msg, update_id=41) + monkeypatch.setattr( + "plugins.platforms.telegram.adapter.cache_audio_from_bytes", + lambda _data, ext: "/tmp/registered-voice.ogg", + ) + + producer = asyncio.create_task( + adapter._handle_media_message(update, SimpleNamespace()) + ) + await asyncio.sleep(0) + merged = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + await producer + + assert merged.message_type is MessageType.VOICE + assert merged.media_urls == ["/tmp/registered-voice.ogg"] + assert adapter._startup_batch_events == {} + assert adapter._media_downloads_in_progress_by_session == {} + + +def test_media_session_key_applies_topic_recovery_and_adapter_profile(): + source = _source() + source.profile = None + source.thread_id = "raw-topic" + adapter = _make_adapter() + adapter._gateway_profile_name = "coder" + adapter._apply_topic_recovery = lambda event: setattr(event.source, "thread_id", "recovered-topic") + event = _photo_event(source) + + key = adapter._event_session_key(event) + + assert key == "agent:coder:telegram:dm:273403055:recovered-topic" + assert event.source.profile == "coder" + assert event.source.thread_id == "recovered-topic" + + +def test_text_batch_key_uses_adapter_profile_and_topic_recovery(): + source = _source() + source.profile = None + source.thread_id = "raw-topic" + adapter = _make_adapter() + adapter._gateway_profile_name = "coder" + adapter._apply_topic_recovery = lambda event: setattr(event.source, "thread_id", "recovered-topic") + + key = adapter._text_batch_key(_forwarded_text_event(source)) + + assert key == "agent:coder:telegram:dm:273403055:recovered-topic" + + +@pytest.mark.asyncio +async def test_base_adapter_busy_guard_uses_secondary_profile_key(): + source = _source() + source.profile = None + adapter = _make_adapter() + adapter._gateway_profile_name = "coder" + adapter._message_handler = AsyncMock() + adapter._busy_session_handler = AsyncMock(return_value=True) + adapter._active_sessions = { + "agent:coder:telegram:dm:273403055:363402": asyncio.Event() + } + adapter._session_tasks = {} + adapter._background_tasks = set() + adapter._heal_stale_session_lock = lambda _key: None + + await adapter.handle_message(_forwarded_photo_event(source)) + + adapter._busy_session_handler.assert_awaited_once() + assert adapter._busy_session_handler.await_args.args[1] == ( + "agent:coder:telegram:dm:273403055:363402" + ) + + +@pytest.mark.asyncio +async def test_gateway_text_only_fast_path_does_not_wait_without_pending_media(monkeypatch): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + sleep = AsyncMock() + monkeypatch.setattr("gateway.run.asyncio.sleep", sleep) + + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + + assert event.message_type == MessageType.TEXT + assert event.text == "Is this a real study?" + assert event.media_urls == [] + sleep.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_gateway_consumes_forwarded_text_from_telegram_debounce(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + adapter._enqueue_text_event(_forwarded_text_event(source)) + + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + + assert "[Forwarded message | From: Alina" in event.text + assert "second forwarded text" in event.text + assert session_key not in adapter._pending_text_batches + assert session_key not in adapter._pending_text_batch_tasks + + +@pytest.mark.asyncio +async def test_gateway_does_not_consume_generic_fifo_document_head(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + queued = _document_event(source) + adapter._pending_messages[session_key] = queued + + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + + assert event.message_type == MessageType.TEXT + assert event.media_urls == [] + assert adapter._pending_messages[session_key] is queued + + +@pytest.mark.asyncio +async def test_gateway_merges_forwarded_text_batch_before_first_model_call(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + adapter._enqueue_text_event(_forwarded_text_event(source)) + + event = await runner._merge_startup_media_followups( + _text_event(source), + source, + session_key, + ) + + assert event.message_type == MessageType.TEXT + assert event.forward_origin is None + assert event.text.startswith("Is this a real study?") + assert "[Forwarded message | From: Alina | Date: 2026-06-14T21:03:26+00:00]" in event.text + assert "second forwarded text" in event.text + assert session_key not in adapter._pending_text_batches + assert session_key not in adapter._pending_text_batch_tasks + + +@pytest.mark.asyncio +async def test_gateway_queues_startup_forwarded_text_batch_without_interrupt_ack(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + runner._running_agents = {session_key: _AGENT_PENDING_SENTINEL} + runner._busy_input_mode = "interrupt" + runner._busy_text_mode = "interrupt" + runner._busy_ack_ts = {} + runner._running_agents_ts = {} + + handled = await runner._handle_active_session_busy_message( + _forwarded_text_event(source), + session_key, + ) + + assert handled is True + assert adapter._startup_batch_events[session_key].text.endswith("second forwarded text") + assert session_key not in adapter._pending_messages + + +@pytest.mark.asyncio +async def test_startup_forward_does_not_absorb_existing_fifo_head(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + runner = _make_runner(adapter) + queued = MessageEvent(text="queued-first", message_type=MessageType.TEXT, source=source) + adapter._pending_messages[session_key] = queued + runner._running_agents = {session_key: _AGENT_PENDING_SENTINEL} + runner._busy_input_mode = "interrupt" + runner._busy_text_mode = "interrupt" + runner._busy_ack_ts = {} + runner._running_agents_ts = {} + + handled = await runner._handle_active_session_busy_message( + _forwarded_photo_event(source), + session_key, + ) + + assert handled is True + assert adapter._pending_messages[session_key] is queued + assert adapter._pending_messages[session_key].text == "queued-first" + assert adapter._startup_batch_events[session_key].media_urls == ["/tmp/forwarded-post.jpg"] + + +def test_startup_slot_preserves_each_forwarded_media_origin_inline(): + source = _source() + session_key = build_session_key(source) + adapter = _make_adapter() + ordinary = MessageEvent( + text="ordinary", + message_type=MessageType.DOCUMENT, + source=source, + media_urls=["/tmp/ordinary.pdf"], + media_types=["application/pdf"], + ) + alice = MessageEvent( + text="alice file", + message_type=MessageType.DOCUMENT, + source=source, + media_urls=["/tmp/alice.pdf"], + media_types=["application/pdf"], + forward_origin={"type": "user", "sender_name": "Alice"}, + ) + bob = MessageEvent( + text="bob file", + message_type=MessageType.DOCUMENT, + source=source, + media_urls=["/tmp/bob.pdf"], + media_types=["application/pdf"], + forward_origin={"type": "user", "sender_name": "Bob"}, + ) + + adapter.queue_startup_batch_event(session_key, ordinary) + adapter.queue_startup_batch_event(session_key, alice) + adapter.queue_startup_batch_event(session_key, bob) + batch = adapter._startup_batch_events[session_key] + + assert batch.forward_origin is None + assert batch.text.count("[Forwarded message | From: Alice]") == 1 + assert batch.text.count("[Forwarded message | From: Bob]") == 1 + assert batch.media_urls == ["/tmp/ordinary.pdf", "/tmp/alice.pdf", "/tmp/bob.pdf"] + + +@pytest.mark.asyncio +async def test_forwarded_context_is_rendered_before_inbound_text(): + source = _source() + adapter = _make_adapter() + runner = _make_runner(adapter) + event = MessageEvent( + text="original text", + message_type=MessageType.TEXT, + source=source, + forward_origin={ + "type": "user", + "sender_name": "Skippy", + "sender_username": "skippy_bot", + "date": "2026-06-14T09:00:00+00:00", + }, + ) + + message_text = await runner._prepare_inbound_message_text( + event=event, + source=source, + history=[], + ) + + assert message_text.startswith( + "[Forwarded message | From: Skippy (@skippy_bot) | Date: 2026-06-14T09:00:00+00:00]" + ) + assert message_text.endswith("original text")