Gateway media and voice delivery bugs and fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #3919 fix: background task media delivery + vision download timeout

  merged to main

  files: cli-config.yaml.example, gateway/platforms/telegram.py, gateway/run.py, hermes_cli/config.py, tools/vision_tools.py, website/docs/user-guide/configuration.md

  Show the diff

  • cli-config.yaml.example

    @@ -324,6 +324,9 @@ compression:
     #   vision:
     #     provider: "auto"
     #     model: ""              # e.g. "google/gemini-2.5-flash", "openai/gpt-4o"
    +#     timeout: 30            # LLM API call timeout (seconds)
    +#     download_timeout: 30   # Image HTTP download timeout (seconds)
    +#                            # Increase for slow connections or self-hosted image servers
     #
     #   # Web page scraping / summarization + browser page text extraction
     #   web_extract:

  • gateway/platforms/telegram.py

    @@ -123,6 +123,7 @@ def __init__(self, config: PlatformConfig):
             super().__init__(config, Platform.TELEGRAM)
             self._app: Optional[Application] = None
             self._bot: Optional[Bot] = None
    +        self._webhook_mode: bool = False
             self._mention_patterns = self._compile_mention_patterns()
             self._reply_to_mode: str = getattr(config, 'reply_to_mode', 'first') or 'first'
             # Buffer rapid/album photo updates so Telegram image bursts are handled
    @@ -458,7 +459,19 @@ async def _setup_dm_topics(self) -> None:
                         self._persist_dm_topic_thread_id(int(chat_id), topic_name, thread_id)
     
         async def connect(self) -> bool:
    -        """Connect to Telegram and start polling for updates."""
    +        """Connect to Telegram via polling or webhook.
    +
    +        By default, uses long polling (outbound connection to Telegram).
    +        If ``TELEGRAM_WEBHOOK_URL`` is set, starts an HTTP webhook server
    +        instead.  Webhook mode is useful for cloud deployments (Fly.io,
    +        Railway) where inbound HTTP can wake a suspended machine.
    +
    +        Env vars for webhook mode::
    +
    +            TELEGRAM_WEBHOOK_URL    Public HTTPS URL (e.g. https://app.fly.dev/telegram)
    +            TELEGRAM_WEBHOOK_PORT   Local listen port (default 8443)
    +            TELEGRAM_WEBHOOK_SECRET Secret token for update verification
    +        """
             if not TELEGRAM_AVAILABLE:
                 logger.error(
                     "[%s] python-telegram-bot not installed. Run: pip install python-telegram-bot",
    @@ -552,27 +565,57 @@ async def connect(self) -> bool:
                         else:
                             raise
                 await self._app.start()
    -            loop = asyncio.get_running_loop()
     
    -            def _polling_error_callback(error: Exception) -> None:
    -                if self._polling_error_task and not self._polling_error_task.done():
    -                    return
    -                if self._looks_like_polling_conflict(error):
    -                    self._polling_error_task = loop.create_task(self._handle_polling_conflict(error))
    -                elif self._looks_like_network_error(error):
    -                    logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error)
    -                    self._polling_error_task = loop.create_task(self._handle_polling_network_error(error))
    -                else:
    -                    logger.error("[%s] Telegram polling error: %s", self.name, error, exc_info=True)
    +            # Decide between webhook and polling mode
    +            webhook_url = os.getenv("TELEGRAM_WEBHOOK_URL", "").strip()
    +
    +            if webhook_url:
    +                # ── Webhook mode ─────────────────────────────────────
    +                # Telegram pushes updates to our HTTP endpoint.  This
    +                # enables cloud platforms (Fly.io, Railway) to auto-wake
    +                # suspended machines on inbound HTTP traffic.
    +                webhook_port = int(os.getenv("TELEGRAM_WEBHOOK_PORT", "8443"))
    +                webhook_secret = [redacted:secret-assignment]"TELEGRAM_WEBHOOK_SECRET", "").strip() or None
    +                from urllib.parse import urlparse
    +                webhook_path = urlparse(webhook_url).path or "/telegram"
    +
    +                await self._app.updater.start_webhook(
    +                    listen="0.0.0.0",
    +                    port=webhook_port,
    +                    url_path=webhook_path,
    +                    webhook_url=webhook_url,
    +                    secret_token=[redacted:secret-assignment]
    +                    allowed_updates=Update.ALL_TYPES,
    +                    drop_pending_updates=True,
    +                )
    +                self._webhook_mode = True
    +                logger.info(
    +                    "[%s] Webhook server listening on 0.0.0.0:%d%s",
    +                    self.name, webhook_port, webhook_path,
    +                )
    +            else:
    +                # ── Polling mode (default) ───────────────────────────
    +                loop = asyncio.get_running_loop()
    +
    +                def _polling_error_callback(error: Exception) -> None:
    +                    if self._polling_error_task and not self._polling_error_task.done():
    +                        return
    +                    if self._looks_like_polling_conflict(error):
    +                        self._polling_error_task = loop.create_task(self._handle_polling_conflict(error))
    +                    elif self._looks_like_network_error(error):
    +                        logger.warning("[%s] Telegram network error, scheduling reconnect: %s", self.name, error)
    +                        self._polling_error_task = loop.create_task(self._handle_polling_network_error(error))
    +                    else:
    +                        logger.error("[%s] Telegram polling error: %s", self.name, error, exc_info=True)
     
    -            # Store reference for retry use in _handle_polling_conflict
    -            self._polling_error_callback_ref = _polling_error_callback
    +                # Store reference for retry use in _handle_polling_conflict
    +                self._polling_error_callback_ref = _polling_error_callback
     
    -            await self._app.updater.start_polling(
    -                allowed_updates=Update.ALL_TYPES,
    -                drop_pending_updates=True,
    -                error_callback=_polling_error_callback,
    -            )
    +                await self._app.updater.start_polling(
    +                    allowed_updates=Update.ALL_TYPES,
    +                    drop_pending_updates=True,
    +                    error_callback=_polling_error_callback,
    +                )
                 
                 # Register bot commands so Telegram shows a hint menu when users type /
                 # List is derived from the central COMMAND_REGISTRY — adding a new
    @@ -592,7 +635,8 @@ def _polling_error_callback(error: Exception) -> None:
                     )
                 
                 self._mark_connected()
    -            logger.info("[%s] Connected and polling for Telegram updates", self.name)
    +            mode = "webhook" if self._webhook_mode else "polling"
    +            logger.info("[%s] Connected to Telegram (%s mode)", self.name, mode)
     
                 # Set up DM topics (Bot API 9.4 — Private Chat Topics)
                 # Runs after connection is established so the bot can call createForumTopic.
    @@ -620,7 +664,7 @@ def _polling_error_callback(error: Exception) -> None:
                 return False
         
         async def disconnect(self) -> None:
    -        """Stop polling, cancel pending album flushes, and disconnect."""
    +        """Stop polling/webhook, cancel pending album flushes, and disconnect."""
             pending_media_group_tasks = list(self._media_group_tasks.values())
             for task in pending_media_group_tasks:
                 task.cancel()

  • gateway/run.py

    @@ -3891,7 +3891,7 @@ def run_sync():
                     # Send media files
                     for media_path in (media_files or []):
                         try:
    -                        await adapter.send_file(
    +                        await adapter.send_document(
                                 chat_id=source.chat_id,
                                 file_path=media_path,
                             )

  • hermes_cli/config.py

    @@ -223,7 +223,8 @@ def ensure_hermes_home():
                 "model": "",           # e.g. "google/gemini-2.5-flash", "gpt-4o"
                 "base_url": "",        # direct OpenAI-compatible endpoint (takes precedence over provider)
                 "api_key": "",         # API key for base_url (falls back to OPENAI_API_KEY)
    -            "timeout": 30,         # seconds — increase for slow local vision models
    +            "timeout": 30,         # seconds — LLM API call timeout; increase for slow local vision models
    +            "download_timeout": 30,  # seconds — image HTTP download timeout; increase for slow connections
             },
             "web_extract": {
                 "provider": "auto",

  • tools/vision_tools.py

    @@ -45,6 +45,28 @@
     
     _debug = DebugSession("vision_tools", env_var="VISION_TOOLS_DEBUG")
     
    +# Configurable HTTP download timeout for _download_image().
    +# Separate from auxiliary.vision.timeout which governs the LLM API call.
    +# Resolution: config.yaml auxiliary.vision.download_timeout → env var → 30s default.
    +def _resolve_download_timeout() -> float:
    +    env_val = os.getenv("HERMES_VISION_DOWNLOAD_TIMEOUT", "").strip()
    +    if env_val:
    +        try:
    +            return float(env_val)
    +        except ValueError:
    +            pass
    +    try:
    +        from hermes_cli.config import load_config
    +        cfg = load_config()
    +        val = cfg.get("auxiliary", {}).get("vision", {}).get("download_timeout")
    +        if val is not None:
    +            return float(val)
    +    except Exception:
    +        pass
    +    return 30.0
    +
    +_VISION_DOWNLOAD_TIMEOUT = _resolve_download_timeout()
    +
     
     def _validate_image_url(url: str) -> bool:
         """
    @@ -146,7 +168,7 @@ async def _ssrf_redirect_guard(response):
                 # Enable follow_redirects to handle image CDNs that redirect (e.g., Imgur, Picsum)
                 # SSRF: event_hooks validates each redirect target against private IP ranges
                 async with httpx.AsyncClient(
    -                timeout=30.0,
    +                timeout=_VISION_DOWNLOAD_TIMEOUT,
                     follow_redirects=True,
                     event_hooks={"response": [_ssrf_redirect_guard]},
                 ) as client:
    @@ -183,6 +205,10 @@ async def _ssrf_redirect_guard(response):
                         exc_info=True,
                     )
         
    +    if last_error is None:
    +        raise RuntimeError(
    +            f"_download_image exited retry loop without attempting (max_retries={max_retries})"
    +        )
         raise last_error
     
     

  • website/docs/user-guide/configuration.md

    @@ -1018,7 +1018,8 @@ auxiliary:
         model: ""                  # e.g. "openai/gpt-4o", "google/gemini-2.5-flash"
         base_url: ""               # Custom OpenAI-compatible endpoint (overrides provider)
         api_key: ""                # API key for base_url (falls back to OPENAI_API_KEY)
    -    timeout: 30                # seconds — increase for slow local vision models
    +    timeout: 30                # seconds — LLM API call; increase for slow local vision models
    +    download_timeout: 30       # seconds — image HTTP download; increase for slow connections
     
       # Web page summarization + browser page text extraction
       web_extract:
    @@ -1042,7 +1043,7 @@ auxiliary:
     ```
     
     :::tip
    -Each auxiliary task has a configurable `timeout` (in seconds). Defaults: vision 30s, web_extract 30s, approval 30s, compression 120s. Increase these if you use slow local models for auxiliary tasks.
    +Each auxiliary task has a configurable `timeout` (in seconds). Defaults: vision 30s, web_extract 30s, approval 30s, compression 120s. Increase these if you use slow local models for auxiliary tasks. Vision also has a separate `download_timeout` (default 30s) for the HTTP image download — increase this for slow connections or self-hosted image servers.
     :::
     
     :::info

• #7948 fix(gateway): preserve queued voice events for STT

  merged to main

  files: gateway/run.py, tests/gateway/test_queue_consumption.py, tests/gateway/test_stt_config.py

  Show the diff

  • gateway/run.py

    @@ -352,19 +352,14 @@ def _build_media_placeholder(event) -> str:
         return "\n".join(parts)
     
     
    -def _dequeue_pending_text(adapter, session_key: str) -> str | None:
    -    """Consume and return the text of a pending queued message.
    +def _dequeue_pending_event(adapter, session_key: str) -> MessageEvent | None:
    +    """Consume and return the full pending event for a session.
     
    -    Preserves media context for captionless photo/document events by
    -    building a placeholder so the message isn't silently dropped.
    +    Queued follow-ups must preserve their media metadata so they can re-enter
    +    the normal image/STT/document preprocessing path instead of being reduced
    +    to a placeholder string.
         """
    -    event = adapter.get_pending_message(session_key)
    -    if not event:
    -        return None
    -    text = event.text
    -    if not text and getattr(event, "media_urls", None):
    -        text = _build_media_placeholder(event)
    -    return text
    +    return adapter.get_pending_message(session_key)
     
     
     def _check_unavailable_skill(command_name: str) -> str | None:
    @@ -2775,6 +2770,162 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     del self._running_agents[_quick_key]
                 self._running_agents_ts.pop(_quick_key, None)
     
    +    async def _prepare_inbound_message_text(
    +        self,
    +        *,
    +        event: MessageEvent,
    +        source: SessionSource,
    +        history: List[Dict[str, Any]],
    +    ) -> Optional[str]:
    +        """Prepare inbound event text for the agent.
    +
    +        Keep the normal inbound path and the queued follow-up path on the same
    +        preprocessing pipeline so sender attribution, image enrichment, STT,
    +        document notes, reply context, and @ references all behave the same.
    +        """
    +        history = history or []
    +        message_text = event.text or ""
    +
    +        _is_shared_thread = (
    +            source.chat_type != "dm"
    +            and source.thread_id
    +            and not getattr(self.config, "thread_sessions_per_user", False)
    +        )
    +        if _is_shared_thread and source.user_name:
    +            message_text = f"[{source.user_name}] {message_text}"
    +
    +        if event.media_urls:
    +            image_paths = []
    +            audio_paths = []
    +            for i, path in enumerate(event.media_urls):
    +                mtype = event.media_types[i] if i < len(event.media_types) else ""
    +                if mtype.startswith("image/") or event.message_type == MessageType.PHOTO:
    +                    image_paths.append(path)
    +                if mtype.startswith("audio/") or event.message_type in (MessageType.VOICE, MessageType.AUDIO):
    +                    audio_paths.append(path)
    +
    +            if image_paths:
    +                message_text = await self._enrich_message_with_vision(
    +                    message_text,
    +                    image_paths,
    +                )
    +
    +            if audio_paths:
    +                message_text = await self._enrich_message_with_transcription(
    +                    message_text,
    +                    audio_paths,
    +                )
    +                _stt_fail_markers = (
    +                    "No STT provider",
    +                    "STT is disabled",
    +                    "can't listen",
    +                    "VOICE_TOOLS_OPENAI_KEY",
    +                )
    +                if any(marker in message_text for marker in _stt_fail_markers):
    +                    _stt_adapter = self.adapters.get(source.platform)
    +                    _stt_meta = {"thread_id": source.thread_id} if source.thread_id else None
    +                    if _stt_adapter:
    +                        try:
    +                            _stt_msg = (
    +                                "🎤 I received your voice message but can't transcribe it — "
    +                                "no speech-to-text provider is configured.\n\n"
    +                                "To enable voice: install faster-whisper "
    +                                "(`pip install faster-whisper` in the Hermes venv) "
    +                                "and set `stt.enabled: true` in config.yaml, "
    +                                "then /restart the gateway."
    +                            )
    +                            if self._has_setup_skill():
    +                                _stt_msg += "\n\nFor full setup instructions, type: `/skill hermes-agent-setup`"
    +                            await _stt_adapter.send(
    +                                source.chat_id,
    +                                _stt_msg,
    +                                metadata=_stt_meta,
    +                            )
    +                        except Exception:
    +                            pass
    +
    +        if event.media_urls and event.message_type == MessageType.DOCUMENT:
    +            import mimetypes as _mimetypes
    +
    +            _TEXT_EXTENSIONS = {".txt", ".md", ".csv", ".log", ".json", ".xml", ".yaml", ".yml", ".toml", ".ini", ".cfg"}
    +            for i, path in enumerate(event.media_urls):
    +                mtype = event.media_types[i] if i < len(event.media_types) else ""
    +                if mtype in ("", "application/octet-stream"):
    +                    import os as _os2
    +
    +                    _ext = _os2.path.splitext(path)[1].lower()
    +                    if _ext in _TEXT_EXTENSIONS:
    +                        mtype = "text/plain"
    +                    else:
    +                        guessed, _ = _mimetypes.guess_type(path)
    +                        if guessed:
    +                            mtype = guessed
    +                if not mtype.startswith(("application/", "text/")):
    +                    continue
    +
    +                import os as _os
    +                import re as _re
    +
    +                basename = _os.path.basename(path)
    +                parts = basename.split("_", 2)
    +                display_name = parts[2] if len(parts) >= 3 else basename
    +                display_name = _re.sub(r'[^\w.\- ]', '_', display_name)
    +
    +                if mtype.startswith("text/"):
    +                    context_note = (
    +                        f"[The user sent a text document: '{display_name}'. "
    +                        f"Its content has been included below. "
    +                        f"The file is also saved at: {path}]"
    +                    )
    +                else:
    +                    context_note = (
    +                        f"[The user sent a document: '{display_name}'. "
    +                        f"The file is saved at: {path}. "
    +                        f"Ask the user what they'd like you to do with it.]"
    +                    )
    +                message_text = f"{context_note}\n\n{message_text}"
    +
    +        if getattr(event, "reply_to_text", None) and event.reply_to_message_id:
    +            reply_snippet = event.reply_to_text[:500]
    +            found_in_history = any(
    +                reply_snippet[:200] in (msg.get("content") or "")
    +                for msg in history
    +                if msg.get("role") in ("assistant", "user", "tool")
    +            )
    +            if not found_in_history:
    +                message_text = f'[Replying to: "{reply_snippet}"]\n\n{message_text}'
    +
    +        if "@" in message_text:
    +            try:
    +                from agent.context_references import preprocess_context_references_async
    +                from agent.model_metadata import get_model_context_length
    +
    +                _msg_cwd = os.environ.get("MESSAGING_CWD", os.path.expanduser("~"))
    +                _msg_ctx_len = get_model_context_length(
    +                    self._model,
    +                    base_url=self._base_url or "",
    +                )
    +                _ctx_result = await preprocess_context_references_async(
    +                    message_text,
    +                    cwd=_msg_cwd,
    +                    context_length=_msg_ctx_len,
    +                    allowed_root=_msg_cwd,
    +                )
    +                if _ctx_result.blocked:
    +                    _adapter = self.adapters.get(source.platform)
    +                    if _adapter:
    +                        await _adapter.send(
    +                            source.chat_id,
    +                            "\n".join(_ctx_result.warnings) or "Context injection refused.",
    +                        )
    +                    return None
    +                if _ctx_result.expanded:
    +                    message_text = _ctx_result.message
    +            except Exception as exc:
    +                logger.debug("@ context reference expansion failed: %s", exc)
    +
    +        return message_text
    +
         async def _handle_message_with_agent(self, event, source, _quick_key: str):
             """Inner handler that runs under the _running_agents sentinel guard."""
             _msg_start_time = time.time()
    @@ -3215,149 +3366,13 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
             # attachments (documents, audio, etc.) are not sent to the vision
             # tool even when they appear in the same message.
             # -----------------------------------------------------------------
    -        message_text = event.text or ""
    -
    -        # -----------------------------------------------------------------
    -        # Sender attribution for shared thread sessions.
    -        #
    -        # When multiple users share a single thread session (the default for
    -        # threads), prefix each message with [sender name] so the agent can
    -        # tell participants apart.  Skip for DMs (single-user by nature) and
    -        # when per-user thread isolation is explicitly enabled.
    -        # -----------------------------------------------------------------
    -        _is_shared_thread = (
    -            source.chat_type != "dm"
    -            and source.thread_id
    -            and not getattr(self.config, "thread_sessions_per_user", False)
    +        message_text = await self._prepare_inbound_message_text(
    +            event=event,
    +            source=source,
    +            history=history,
             )
    -        if _is_shared_thread and source.user_name:
    -            message_text = f"[{source.user_name}] {message_text}"
    -
    -        if event.media_urls:
    -            image_paths = []
    -            for i, path in enumerate(event.media_urls):
    -                # Check media_types if available; otherwise infer from message type
    -                mtype = event.media_types[i] if i < len(event.media_types) else ""
    -                is_image = (
    -                    mtype.startswith("image/")
    -                    or event.message_type == MessageType.PHOTO
    -                )
    -                if is_image:
    -                    image_paths.append(path)
    -            if image_paths:
    -                message_text = await self._enrich_message_with_vision(
    -                    message_text, image_paths
    -                )
    -        
    -        # -----------------------------------------------------------------
    -        # Auto-transcribe voice/audio messages sent by the user
    -        # -----------------------------------------------------------------
    -        if event.media_urls:
    -            audio_paths = []
    -            for i, path in enumerate(event.media_urls):
    -                mtype = event.media_types[i] if i < len(event.media_types) else ""
    -                is_audio = (
    -                    mtype.startswith("audio/")
    -                    or event.message_type in (MessageType.VOICE, MessageType.AUDIO)
    -                )
    -                if is_audio:
    -                    audio_paths.append(path)
    -            if audio_paths:
    -                message_text = await self._enrich_message_with_transcription(
    -                    message_text, audio_paths
    -                )
    -                # If STT failed, send a direct message to the user so they
    -                # know voice isn't configured — don't rely on the agent to
    -                # relay the error clearly.
    -                _stt_fail_markers = (
    -                    "No STT provider",
    -                    "STT is disabled",
    -                    "can't listen",
    -                    "VOICE_TOOLS_OPENAI_KEY",
    -                )
    -                if any(m in message_text for m in _stt_fail_markers):
    -                    _stt_adapter = self.adapters.get(source.platform)
    -                    _stt_meta = {"thread_id": source.thread_id} if source.thread_id else None
    -                    if _stt_adapter:
    -                        try:
    -                            _stt_msg = (
    -                                "🎤 I received your voice message but can't transcribe it — "
    -                                "no speech-to-text provider is configured.\n\n"
    -                                "To enable voice: install faster-whisper "
    -                                "(`pip install faster-whisper` in the Hermes venv) "
    -                                "and set `stt.enabled: true` in config.yaml, "
    -                                "then /restart the gateway."
    -                            )
    -                            # Point to setup skill if it's installed
    -                            if self._has_setup_skill():
    -                                _stt_msg += "\n\nFor full setup instructions, type: `/skill hermes-agent-setup`"
    -                            await _stt_adapter.send(
    -                                source.chat_id, _stt_msg,
    -                                metadata=_stt_meta,
    -                            )
    -                        except Exception:
    -                            pass
    -
    -        # -----------------------------------------------------------------
    -        # Enrich document messages with context notes for the agent
    -        # -----------------------------------------------------------------
    -        if event.media_urls and event.message_type == MessageType.DOCUMENT:
    -            import mimetypes as _mimetypes
    -            _TEXT_EXTENSIONS = {".txt", ".md", ".csv", ".log", ".json", ".xml", ".yaml", ".yml", ".toml", ".ini", ".cfg"}
    -            for i, path in enumerate(event.media_urls):
    -                mtype = event.media_types[i] if i < len(event.media_types) else ""
    -                # Fall back to extension-based detection when MIME type is unreliable.
    -                if mtype in ("", "application/octet-stream"):
    -                    import os as _os2
    -                    _ext = _os2.path.splitext(path)[1].lower()
    -                    if _ext in _TEXT_EXTENSIONS:
    -                        mtype = "text/plain"
    -                    else:
    -                        guessed, _ = _mimetypes.guess_type(path)
    -                        if guessed:
    -                            mtype = guessed
    -                if not mtype.startswith(("application/", "text/")):
    -                    continue
    -                # Extract display filename by stripping the doc_{uuid12}_ prefix
    -                import os as _os
    -                basename = _os.path.basename(path)
    -                # Format: doc_<12hex>_<original_filename>
    -                parts = basename.split("_", 2)
    -                display_name = parts[2] if len(parts) >= 3 else basename
    -                # Sanitize to prevent prompt injection via filenames
    -                import re as _re
    -                display_name = _re.sub(r'[^\w.\- ]', '_', display_name)
    -
    -                if mtype.startswith("text/"):
    -                    context_note = (
    -                        f"[The user sent a text document: '{display_name}'. "
    -                        f"Its content has been included below. "
    -                        f"The file is also saved at: {path}]"
    -                    )
    -                else:
    -                    context_note = (
    -                        f"[The user sent a document: '{display_name}'. "
    -                        f"The file is saved at: {path}. "
    -                        f"Ask the user what they'd like you to do with it.]"
    -                    )
    -                message_text = f"{context_note}\n\n{message_text}"
    -
    -        # -----------------------------------------------------------------
    -        # Inject reply context when user replies to a message not in history.
    -        # Telegram (and other platforms) let users reply to specific messages,
    -        # but if the quoted message is from a previous session, cron delivery,
    -        # or background task, the agent has no context about what's being
    -        # referenced. Prepend the quoted text so the agent understands. (#1594)
    -        # -----------------------------------------------------------------
    -        if getattr(event, 'reply_to_text', None) and event.reply_to_message_id:
    -            reply_snippet = event.reply_to_text[:500]
    -            found_in_history = any(
    -                reply_snippet[:200] in (msg.get("content") or "")
    -                for msg in history
    -                if msg.get("role") in ("assistant", "user", "tool")
    -            )
    -            if not found_in_history:
    -                message_text = f'[Replying to: "{reply_snippet}"]\n\n{message_text}'
    +        if message_text is None:
    +            return
     
             try:
                 # Emit agent:start hook
    @@ -3369,30 +3384,6 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str):
                 }
                 await self.hooks.emit("agent:start", hook_ctx)
     
    -            # Expand @ context references (@file:, @folder:, @diff, etc.)
    -            if "@" in message_text:
    -                try:
    -                    from agent.context_references import preprocess_context_references_async
    -                    from agent.model_metadata import get_model_context_length
    -                    _msg_cwd = os.environ.get("MESSAGING_CWD", os.path.expanduser("~"))
    -                    _msg_ctx_len = get_model_context_length(
    -                        self._model, base_url=self._base_url or "")
    -                    _ctx_result = await preprocess_context_references_async(
    -                        message_text, cwd=_msg_cwd,
    -                        context_length=_msg_ctx_len, allowed_root=_msg_cwd)
    -                    if _ctx_result.blocked:
    -                        _adapter = self.adapters.get(source.platform)
    -                        if _adapter:
    -                            await _adapter.send(
    -                                source.chat_id,
    -                                "\n".join(_ctx_result.warnings) or "Context injection refused.",
    -                            )
    -                        return
    -                    if _ctx_result.expanded:
    -                        message_text = _ctx_result.message
    -                except Exception as exc:
    -                    logger.debug("@ context reference expansion failed: %s", exc)
    -
                 # Run the agent
                 agent_result = await self._run_agent(
                     message=message_text,
    @@ -8057,17 +8048,16 @@ async def _notify_long_running():
                 
                 # Get pending message from adapter.
                 # Use session_key (not source.chat_id) to match adapter's storage keys.
    +            pending_event = None
                 pending = None
                 if result and adapter and session_key:
    -                if result.get("interrupted"):
    -                    pending = _dequeue_pending_text(adapter, session_key)
    -                    if not pending and result.get("interrupt_message"):
    -                        pending = result.get("interrupt_message")
    -                else:
    -                    pending = _dequeue_pending_text(adapter, session_key)
    -                    if pending:
    -                        logger.debug("Processing queued message after agent completion: '%s...'", pending[:40])
    -            
    +                pending_event = _dequeue_pending_event(adapter, session_key)
    +                if result.get("interrupted") and not pending_event and result.get("interrupt_message"):
    +                    pending = result.get("interrupt_message")
    +                elif pending_event:
    +                    pending = pending_event.text or _build_media_placeholder(pending_event)
    +                    logger.debug("Processing queued message after agent completion: '%s...'", pending[:40])
    +
                 # Safety net: if the pending text is a slash command (e.g. "/stop",
                 # "/new"), discard it — commands should never be passed to the agent
                 # as user input.  The primary fix is in base.py (commands bypass the
    @@ -8085,27 +8075,29 @@ async def _notify_long_running():
                                     "commands must not be passed as agent input",
                                     _pending_cmd_word,
                                 )
    +                            pending_event = None
                                 pending = None
                         except Exception:
                             pass
     
    -            if self._draining and pending:
    +            if self._draining and (pending_event or pending):
                     logger.info(
                         "Discarding pending follow-up for session %s during gateway %s",
                         session_key[:20] if session_key else "?",
                         self._status_action_label(),
                     )
    +                pending_event = None
                     pending = None
     
    -            if pending:
    +            if pending_event or pending:
                     logger.debug("Processing pending message: '%s...'", pending[:40])
    -                
    +
                     # Clear the adapter's interrupt event so the next _run_agent call
                     # doesn't immediately re-trigger the interrupt before the new agent
                     # even makes its first API call (this was causing an infinite loop).
                     if adapter and hasattr(adapter, '_active_sessions') and session_key and session_key in adapter._active_sessions:
                         adapter._active_sessions[session_key].clear()
    -                
    +
                     # Cap recursion depth to prevent resource exhaustion when the
                     # user sends multiple messages while the agent keeps failing. (#816)
                     if _interrupt_depth >= self._MAX_INTERRUPT_DEPTH:
    @@ -8114,9 +8106,10 @@ async def _notify_long_running():
                             "queueing message instead of recursing.",
                             _interrupt_depth, session_key,
                         )
    -                    # Queue the pending message for normal processing on next turn
                         adapter = self.adapters.get(source.platform)
    -                    if adapter and hasattr(adapter, 'queue_message'):
    +                    if adapter and pending_event:
    +                        merge_pending_message_event(adapter._pending_messages, session_key, pending_event)
    +                    elif adapter and hasattr(adapter, 'queue_message'):
                             adapter.queue_message(session_key, pending)
                         return result_holder[0] or {"final_response": response, "messages": history}
     
    @@ -8138,16 +8131,30 @@ async def _notify_long_running():
                     # interrupted." is just noise; the user already knows they sent a
                     # new message).
     
    -                # Process the pending message with updated history
                     updated_history = result.get("messages", history)
    +                next_source = source
    +                next_message = pending
    +                next_message_id = None
    +                if pending_event is not None:
    +                    next_source = getattr(pending_event, "source", None) or source
    +                    next_message = await self._prepare_inbound_message_text(
    +                        event=pending_event,
    +                        source=next_source,
    +                        history=updated_history,
    +                    )
    +                    if next_message is None:
    +                        return result
    +                    next_message_id = getattr(pending_event, "message_id", None)
    +
                     return await self._run_agent(
    -                    message=pending,
    +                    message=next_message,
                         context_prompt=context_prompt,
                         history=updated_history,
    -                    source=source,
    +                    source=next_source,
                         session_id=session_id,
                         session_key=session_key,
                         _interrupt_depth=_interrupt_depth + 1,
    +                    event_message_id=next_message_id,
                     )
             finally:
                 # Stop progress sender, interrupt monitor, and notification task

  • tests/gateway/test_queue_consumption.py

    @@ -10,6 +10,7 @@
     
     import pytest
     
    +from gateway.run import _dequeue_pending_event
     from gateway.platforms.base import (
         BasePlatformAdapter,
         MessageEvent,
    @@ -79,6 +80,26 @@ def test_get_pending_message_consumes_and_clears(self):
             # Should be consumed (cleared)
             assert adapter.get_pending_message(session_key) is None
     
    +    def test_dequeue_pending_event_preserves_voice_media_metadata(self):
    +        adapter = _StubAdapter()
    +        session_key = "telegram:user:voice"
    +        event = MessageEvent(
    +            text="",
    +            message_type=MessageType.VOICE,
    +            source=MagicMock(chat_id="123", platform=Platform.TELEGRAM),
    +            message_id="voice-q1",
    +            media_urls=["/tmp/voice.ogg"],
    +            media_types=["audio/ogg"],
    +        )
    +        adapter._pending_messages[session_key] = event
    +
    +        retrieved = _dequeue_pending_event(adapter, session_key)
    +
    +        assert retrieved is event
    +        assert retrieved.media_urls == ["/tmp/voice.ogg"]
    +        assert retrieved.media_types == ["audio/ogg"]
    +        assert adapter.get_pending_message(session_key) is None
    +
         def test_queue_does_not_set_interrupt_event(self):
             """The whole point of /queue — no interrupt signal."""
             adapter = _StubAdapter()

  • tests/gateway/test_stt_config.py

    @@ -6,7 +6,9 @@
     import pytest
     import yaml
     
    -from gateway.config import GatewayConfig, load_gateway_config
    +from gateway.config import GatewayConfig, Platform, load_gateway_config
    +from gateway.platforms.base import MessageEvent, MessageType
    +from gateway.session import SessionSource
     
     
     def test_gateway_config_stt_disabled_from_dict_nested():
    @@ -69,3 +71,46 @@ async def test_enrich_message_with_transcription_avoids_bogus_no_provider_messag
         assert "No STT provider is configured" not in result
         assert "trouble transcribing" in result
         assert "caption" in result
    +
    +
    [redacted:email]
    +async def test_prepare_inbound_message_text_transcribes_queued_voice_event():
    +    from gateway.run import GatewayRunner
    +
    +    runner = GatewayRunner.__new__(GatewayRunner)
    +    runner.config = GatewayConfig(stt_enabled=True)
    +    runner.adapters = {}
    +    runner._model = "test-model"
    +    runner._base_url = ""
    +    runner._has_setup_skill = lambda: False
    +
    +    source = SessionSource(
    +        platform=Platform.TELEGRAM,
    +        chat_id="123",
    +        chat_type="dm",
    +    )
    +    event = MessageEvent(
    +        text="",
    +        message_type=MessageType.VOICE,
    +        source=source,
    +        media_urls=["/tmp/queued-voice.ogg"],
    +        media_types=["audio/ogg"],
    +    )
    +
    +    with patch(
    +        "tools.transcription_tools.transcribe_audio",
    +        return_value={
    +            "success": True,
    +            "transcript": "queued voice transcript",
    +            "provider": "local_command",
    +        },
    +    ):
    +        result = await runner._prepare_inbound_message_text(
    +            event=event,
    +            source=source,
    +            history=[],
    +        )
    +
    +    assert result is not None
    +    assert "queued voice transcript" in result
    +    assert "voice message" in result.lower()

• #82162 fix(gateway): preserve MEDIA delivery on queued follow-up responses (salvage #71031)

  merged to main

  files: gateway/run.py, tests/gateway/test_run_progress_topics.py, tests/gateway/test_tts_media_routing.py

• #2339 fix(gateway): pass message_thread_id in send_image_file, send_document, send_video

  merged to main

  files: gateway/platforms/telegram.py, tests/gateway/test_send_image_file.py, tests/gateway/test_telegram_documents.py